app.repository package

Submodules

app.repository.base_repository module

class app.repository.base_repository.BaseRepository(session_factory: Callable[[...], AbstractContextManager[Session]], model)

Bases: object

Base repository providing common CRUD operations.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class.

async create(schema)

Creates a new record.

Parameters:

schema – The schema containing the record data.

Returns:

The created record.

Return type:

object

delete_by_id(id: int)

Deletes a record by its ID.

Parameters:

id (int) – The record ID.

Returns:

None

read_by_column(column: str, value: str, eager=False, only_one=True, not_found_raise_exception=True, not_found_message='Not found {column} : {value}')

Reads records by a specified column and value.

Parameters:

• column (str) – The column name.

• value (str) – The value to filter by.

• eager (bool) – Whether to use eager loading.

• only_one (bool) – Whether to return only one record.

• not_found_raise_exception (bool) – Whether to raise an exception if the record is not found.

• not_found_message (str) – The message for the not found exception.

Returns:

The record(s) if found, otherwise None or raises

an exception.

Return type:

object or list

read_by_id(id: int, eager=False, not_found_raise_exception=True, not_found_message='Not found id : {id}')

Reads a record by its ID.

Parameters:

• id (int) – The record ID.

• eager (bool) – Whether to use eager loading.

• not_found_raise_exception (bool) – Whether to raise an exception if the record is not found.

• not_found_message (str) – The message for the not found exception.

Returns:

The record if found, otherwise None or raises an exception.

Return type:

object

read_by_options(schema, eager=False)

Reads records by specified options.

Parameters:

• schema – The schema containing query options.

• eager (bool) – Whether to use eager loading.

Returns:

Query results and search options.

Return type:

dict

update(id: int, schema)

Updates a record by its ID.

Parameters:

• id (int) – The record ID.

• schema – The schema containing the updated data.

Returns:

The updated record.

Return type:

object

update_attr(id: int, column: str, value)

Updates a specific attribute of a record by its ID.

Parameters:

• id (int) – The record ID.

• column (str) – The column name.

• value – The new value of the attribute.

Returns:

The updated record.

Return type:

object

whole_update(id: int, schema)

Replaces a record entirely by its ID.

Parameters:

• id (int) – The record ID.

• schema – The schema containing the new data.

Returns:

The updated record.

Return type:

object

app.repository.game_params_repository module

class app.repository.game_params_repository.GameParamsRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.game_params.GamesParams'>)

Bases: BaseRepository

Repository class for game parameters.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for game parameters.

patch_game_params_by_id(id: str, schema)

Updates game parameters by their ID.

Parameters:

• id (str) – The game parameter ID.

• schema – The schema containing the updated data.

Returns:

The updated game parameters.

Return type:

object

Raises:

NotFoundError – If the game parameters are not found.

app.repository.game_repository module

class app.repository.game_repository.GameRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.games.Games'>, model_tasks=<class 'app.model.tasks.Tasks'>, model_game_params=<class 'app.model.game_params.GamesParams'>, model_tasks_params=<class 'app.model.task_params.TasksParams'>, model_user_points=<class 'app.model.user_points.UserPoints'>)

Bases: BaseRepository

Repository class for games.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for games.

model_tasks

SQLAlchemy model class for tasks.

model_game_params

SQLAlchemy model class for game parameters.

delete_game_by_id(game_id: str)

Deletes a game by its ID , it’s delete all games related with gameId in

games, game_params, tasks and tasks_params.

Parameters:

game_id (str) – The game ID.

Raises:

NotFoundError – If the game is not found.

get_all_games(schema, api_key=None)

Retrieves all games based on the provided schema.

Parameters:

schema – The schema for filtering the games.

Returns:

A result set containing the games and search

options.

Return type:

FindGameResult

get_game_by_id(id: str)

Retrieves a game by its ID.

Parameters:

id (str) – The game ID.

Returns:

The game details.

Return type:

BaseGameResult

patch_game_by_id(gameId: str, schema)

Updates a game by its ID using the provided schema.

Parameters:

• gameId (str) – The game ID.

• schema – The schema representing the updated data.

Returns:

The updated game details.

Return type:

BaseGameResult

Raises:

• NotFoundError – If the game is not found.

• DuplicatedError – If a duplicated error occurs during update.

app.repository.task_params_repository module

class app.repository.task_params_repository.TaskParamsRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.task_params.TasksParams'>)

Bases: BaseRepository

Repository class for task parameters.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for task parameters.

app.repository.task_repository module

class app.repository.task_repository.TaskRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.tasks.Tasks'>)

Bases: BaseRepository

Repository class for tasks.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for tasks.

get_points_and_users_by_taskId(taskId)

Retrieves points and users associated with a task ID.

Parameters:

taskId (int) – The task ID.

Returns:

The task if found, otherwise raises NotFoundError.

Return type:

object

read_by_gameId(schema, eager=False)

Reads tasks by game ID based on the provided schema.

Parameters:

• schema – The schema containing query options.

• eager (bool) – Whether to use eager loading.

Returns:

Query results and search options.

Return type:

dict

read_by_gameId_and_externalTaskId(gameId: int, externalTaskId: str)

Reads a task by game ID and external task ID.

Parameters:

• gameId (int) – The game ID.

• externalTaskId (str) – The external task ID.

Returns:

The task if found, otherwise None.

Return type:

object

app.repository.user_points_repository module

class app.repository.user_points_repository.UserPointsRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.user_points.UserPoints'>)

Bases: BaseRepository

Repository class for user points.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for user points.

task_repository

Repository instance for tasks.

Type:

BaseRepository

user_repository

Repository instance for users.

Type:

BaseRepository

count_measurements_by_external_task_id(external_task_id)

Retrieves the total number of measurements by external task ID.

Parameters:

external_task_id (str) – The external task ID.

Returns:

The total number of measurements completed for the task.

Return type:

int

count_personal_records_by_external_game_id(externalGameId, externalUserId)

Retrieves the total number of personal records by external game ID.

Parameters:

• externalGameId (str) – The external game ID.

• externalUserId (str) – The external user ID.

Returns:

The total number of personal records.

Return type:

int

get_all_UserPoints_by_gameId(gameId)

Retrieves all user points associated with a game ID.

Parameters:

gameId (int) – The game ID.

Returns:

A list of user points grouped by task and user.

Return type:

list

get_all_UserPoints_by_taskId(taskId)

Retrieves all user points for a specific task.

Parameters:

taskId (str) – The task ID.

Returns:

A list of user points.

Return type:

list

get_all_UserPoints_by_taskId_with_details(taskId)

Retrieves all user points for a specific task with details.

Parameters:

taskId (str) – The task ID.

Returns:

A list of user points with detailed information.

Return type:

list

get_avg_time_between_tasks_by_user_and_game_task(externalGameId, externalTaskId, externalUserId)

Retrieves the average time between tasks for a user and game task.

Parameters:

• externalGameId (str) – The external game ID.

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

Returns:

The average time between tasks.

Return type:

float

get_avg_time_between_tasks_for_all_users(externalGameId, externalTaskId)

Retrieves the average time between tasks for all users for a game task.

Parameters:

• externalGameId (str) – The external game ID.

• externalTaskId (str) – The external task ID.

Returns:

The average time between tasks for all users.

Return type:

float

get_first_user_points_in_external_task_id_by_user_id(externalTaskId, externalUserId)

Retrieves the first user points in an external task by user ID.

Parameters:

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

Returns:

The first user points in the task.

Return type:

UserPoints

get_global_avg_by_external_game_id(externalGameId)

Retrieves the global average time using data[“minutes”] for a given

external game ID.

Parameters:

externalGameId (str) – The external game ID.

Returns:

The global average time, or -1 if no valid data exists.

Return type:

float

get_global_calculation()

Calculates and retrieves a global performance metric.

Returns:

The calculated global performance metric.

Return type:

float

get_individual_calculation(userId)

Calculates and retrieves an individual performance metric for a user.

Parameters:

userId (str) – The user ID.

Returns:

The calculated individual performance metric.

Return type:

float

get_last_window_time_diff(externalTaskId, externalUserId)

Retrieves the time difference between the last two measurements by

a user for a task.

Parameters:

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

Returns:

The time difference between the last two measurements.

Return type:

float

get_new_last_window_time_diff(externalTaskId, externalUserId, externalGameId)

Retrieves the time difference between the last measurement and current

time for a user for a task in a game.

Parameters:

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

• externalGameId (str) – The external game ID.

Returns:

The time difference between the last measurement and

current time.

Return type:

float

get_personal_avg_by_external_game_id(externalGameId, externalUserId)

Retrieves the personal average time using data[“minutes”] for a given

external game ID and user ID.

Parameters:

• externalGameId (str) – The external game ID.

• externalUserId (str) – The external user ID.

Returns:

The personal average time, or -1 if no valid data exists.

Return type:

float

get_points_and_users_by_taskId(taskId)

Retrieves points and users associated with a task ID.

Parameters:

taskId (int) – The task ID.

Returns:

A list of user points with user information.

Return type:

list

get_start_time_for_last_task(userId)

Retrieves the start time of the last task completed by a user.

Parameters:

userId (str) – The user ID.

Returns:

The start time of the last task.

Return type:

datetime

get_task_and_sum_points_by_userId(userId)

Retrieves tasks and the sum of points for a user by their user ID.

Parameters:

userId (str) – The user ID.

Returns:

A list of tasks with the sum of points.

Return type:

list

get_task_by_externalUserId(externalUserId)

Retrieves tasks associated with a user by their external user ID.

Parameters:

externalUserId (str) – The external user ID.

Returns:

A list of tasks.

Return type:

list

get_time_taken_for_last_task(userId)

Retrieves the time taken by a user to complete the last task.

Parameters:

userId (str) – The user ID.

Returns:

The time taken to complete the last task.

Return type:

datetime

get_user_measurement_count(userId)

Retrieves the total number of measurements (tasks completed) by a

specific user.

Parameters:

userId (str) – The user ID.

Returns:

The total number of measurements completed by the user.

Return type:

int

get_user_task_measurements(externalTaskId, externalUserId)

Retrieves measurements for a user and task.

Parameters:

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

Returns:

A list of measurements for the user and task.

Return type:

list

get_user_task_measurements_count(externalTaskId, externalUserId)

Retrieves the total number of measurements by user and task.

Parameters:

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

Returns:

The total number of measurements by user and task.

Return type:

int

get_user_task_measurements_count_the_last_seconds(externalTaskId, externalUserId, seconds)

Retrieves the total number of measurements by user and task in the last

n seconds.

Parameters:

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

• seconds (int) – The number of seconds to consider.

Returns:

The total number of measurements by user and task in the last

n seconds.

Return type:

int

user_has_record_before_in_externalTaskId_last_min(externalTaskId, externalUserId, minutes)

Checks if a user has at least one record before in an external task.

Parameters:

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

• minutes (int) – The number of minutes to consider.

Returns:

True if the user has one record before, False otherwise.

Return type:

bool

app.repository.user_repository module

class app.repository.user_repository.UserRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.users.Users'>)

Bases: BaseRepository

Repository class for users.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for users.

create_user_by_externalUserId(externalUserId: str)

Creates a new user with the provided external user ID.

Parameters:

externalUserId (str) – The external user ID.

Returns:

The created user.

Return type:

Users

app.repository.wallet_repository module

class app.repository.wallet_repository.WalletRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.wallet.Wallet'>)

Bases: BaseRepository

Repository class for wallets.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for wallets.

app.repository.wallet_transaction_repository module

class app.repository.wallet_transaction_repository.WalletTransactionRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.wallet_transactions.WalletTransactions'>)

Bases: BaseRepository

Repository class for wallet transactions.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for wallet transactions.

Module contents

class app.repository.ApiKeyRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.api_key.ApiKey'>)

Bases: BaseRepository

Repository class for API keys.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for API keys.

read_all(page: int = 1, page_size: int = 100)

Reads all API keys. Order by created_at.

Parameters:

• page (int) – The page number.

• page_size (int) – The number of items per page.

Returns:

All API keys in the database.

Return type:

List[ApiKey]

class app.repository.ApiRequestsRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.api_requests.ApiRequests'>)

Bases: BaseRepository

Repository class for API Requests.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for API Requests.

class app.repository.BaseRepository(session_factory: Callable[[...], AbstractContextManager[Session]], model)

Bases: object

Base repository providing common CRUD operations.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class.

async create(schema)

Creates a new record.

Parameters:

schema – The schema containing the record data.

Returns:

The created record.

Return type:

object

delete_by_id(id: int)

Deletes a record by its ID.

Parameters:

id (int) – The record ID.

Returns:

None

read_by_column(column: str, value: str, eager=False, only_one=True, not_found_raise_exception=True, not_found_message='Not found {column} : {value}')

Reads records by a specified column and value.

Parameters:

• column (str) – The column name.

• value (str) – The value to filter by.

• eager (bool) – Whether to use eager loading.

• only_one (bool) – Whether to return only one record.

• not_found_raise_exception (bool) – Whether to raise an exception if the record is not found.

• not_found_message (str) – The message for the not found exception.

Returns:

The record(s) if found, otherwise None or raises

an exception.

Return type:

object or list

read_by_id(id: int, eager=False, not_found_raise_exception=True, not_found_message='Not found id : {id}')

Reads a record by its ID.

Parameters:

• id (int) – The record ID.

• eager (bool) – Whether to use eager loading.

• not_found_raise_exception (bool) – Whether to raise an exception if the record is not found.

• not_found_message (str) – The message for the not found exception.

Returns:

The record if found, otherwise None or raises an exception.

Return type:

object

read_by_options(schema, eager=False)

Reads records by specified options.

Parameters:

• schema – The schema containing query options.

• eager (bool) – Whether to use eager loading.

Returns:

Query results and search options.

Return type:

dict

update(id: int, schema)

Updates a record by its ID.

Parameters:

• id (int) – The record ID.

• schema – The schema containing the updated data.

Returns:

The updated record.

Return type:

object

update_attr(id: int, column: str, value)

Updates a specific attribute of a record by its ID.

Parameters:

• id (int) – The record ID.

• column (str) – The column name.

• value – The new value of the attribute.

Returns:

The updated record.

Return type:

object

whole_update(id: int, schema)

Replaces a record entirely by its ID.

Parameters:

• id (int) – The record ID.

• schema – The schema containing the new data.

Returns:

The updated record.

Return type:

object

class app.repository.DashboardRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model_games=<class 'app.model.games.Games'>, model_tasks=<class 'app.model.tasks.Tasks'>, model_users=<class 'app.model.users.Users'>, model_logs=<class 'app.model.logs.Logs'>, model_user_points=<class 'app.model.user_points.UserPoints'>, model_user_actions=<class 'app.model.user_actions.UserActions'>)

Bases: BaseRepository

Repository class for API keys and dashboard metrics.

get_dashboard_summary(start_date, end_date, group_by)

Retrieves the dashboard summary.

Parameters:

• start_date – The start date for the summary.

• end_date – The end date for the summary.

• group_by – The group by for the summary (e.g. day, week, month).

Returns:

The dashboard summary.

Return type:

Dict[str, Any]

get_dashboard_summary_logs(start_date, end_date, group_by)

Retrieves the dashboard summary logs.

Parameters:

• start_date – The start date for the summary.

• end_date – The end date for the summary.

• group_by – The group by for the summary (e.g. day, week, month).

Returns:

The dashboard summary logs.

Return type:

Dict[str, Any]

process_query(query, start_date=None, end_date=None, group_by_column=None)

Processes the query by filtering and grouping the results.

Parameters:

• query – The query to process.

• start_date – The start date for the query.

• end_date – The end date for the query.

• group_by_column – The column to group by.

Returns:

The processed query.

Return type:

Any

class app.repository.GameParamsRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.game_params.GamesParams'>)

Bases: BaseRepository

Repository class for game parameters.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for game parameters.

patch_game_params_by_id(id: str, schema)

Updates game parameters by their ID.

Parameters:

• id (str) – The game parameter ID.

• schema – The schema containing the updated data.

Returns:

The updated game parameters.

Return type:

object

Raises:

NotFoundError – If the game parameters are not found.

class app.repository.GameRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.games.Games'>, model_tasks=<class 'app.model.tasks.Tasks'>, model_game_params=<class 'app.model.game_params.GamesParams'>, model_tasks_params=<class 'app.model.task_params.TasksParams'>, model_user_points=<class 'app.model.user_points.UserPoints'>)

Bases: BaseRepository

Repository class for games.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for games.

model_tasks

SQLAlchemy model class for tasks.

model_game_params

SQLAlchemy model class for game parameters.

delete_game_by_id(game_id: str)

Deletes a game by its ID , it’s delete all games related with gameId in

games, game_params, tasks and tasks_params.

Parameters:

game_id (str) – The game ID.

Raises:

NotFoundError – If the game is not found.

get_all_games(schema, api_key=None)

Retrieves all games based on the provided schema.

Parameters:

schema – The schema for filtering the games.

Returns:

A result set containing the games and search

options.

Return type:

FindGameResult

get_game_by_id(id: str)

Retrieves a game by its ID.

Parameters:

id (str) – The game ID.

Returns:

The game details.

Return type:

BaseGameResult

patch_game_by_id(gameId: str, schema)

Updates a game by its ID using the provided schema.

Parameters:

• gameId (str) – The game ID.

• schema – The schema representing the updated data.

Returns:

The updated game details.

Return type:

BaseGameResult

Raises:

• NotFoundError – If the game is not found.

• DuplicatedError – If a duplicated error occurs during update.

class app.repository.KpiMetricsRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.kpi_metrics.KpiMetrics'>)

Bases: BaseRepository

Repository class for KPI metrics.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for KPI metrics.

class app.repository.LogsRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.logs.Logs'>)

Bases: BaseRepository

Repository class for Logs.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for Logs.

class app.repository.OAuthUsersRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.oauth_users.OAuthUsers'>)

Bases: BaseRepository

Repository class for OAuth users.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for OAuth users.

get_user_by_sub(sub: str)

Get a user by their sub.

Parameters:

sub – The sub of the user to get (provider_user_id)

Returns:

The user with the provided sub.

class app.repository.TaskParamsRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.task_params.TasksParams'>)

Bases: BaseRepository

Repository class for task parameters.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for task parameters.

class app.repository.TaskRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.tasks.Tasks'>)

Bases: BaseRepository

Repository class for tasks.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for tasks.

get_points_and_users_by_taskId(taskId)

Retrieves points and users associated with a task ID.

Parameters:

taskId (int) – The task ID.

Returns:

The task if found, otherwise raises NotFoundError.

Return type:

object

read_by_gameId(schema, eager=False)

Reads tasks by game ID based on the provided schema.

Parameters:

• schema – The schema containing query options.

• eager (bool) – Whether to use eager loading.

Returns:

Query results and search options.

Return type:

dict

read_by_gameId_and_externalTaskId(gameId: int, externalTaskId: str)

Reads a task by game ID and external task ID.

Parameters:

• gameId (int) – The game ID.

• externalTaskId (str) – The external task ID.

Returns:

The task if found, otherwise None.

Return type:

object

class app.repository.UptimeLogsRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.uptime_logs.UptimeLogs'>)

Bases: BaseRepository

Repository class for Uptime Logs.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for Uptime Logs.

class app.repository.UserActionsRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.user_actions.UserActions'>)

Bases: BaseRepository

Repository class for user points.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for user points.

add_action_in_task(user_id: str, task_id: str, action: AddActionDidByUserInTask)

Add action in task for user.

Parameters:

• user_id (str) – The user ID.

• task_id (str) – The task ID.

• action (AddActionDidByUserInTask) – The action to add.

Returns:

The added action in task for user.

Return type:

object

class app.repository.UserInteractionsRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.user_interactions.UserInteractions'>)

Bases: BaseRepository

Repository class for User interactions.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for User interactions.

class app.repository.UserPointsRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.user_points.UserPoints'>)

Bases: BaseRepository

Repository class for user points.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for user points.

task_repository

Repository instance for tasks.

Type:

BaseRepository

user_repository

Repository instance for users.

Type:

BaseRepository

count_measurements_by_external_task_id(external_task_id)

Retrieves the total number of measurements by external task ID.

Parameters:

external_task_id (str) – The external task ID.

Returns:

The total number of measurements completed for the task.

Return type:

int

count_personal_records_by_external_game_id(externalGameId, externalUserId)

Retrieves the total number of personal records by external game ID.

Parameters:

• externalGameId (str) – The external game ID.

• externalUserId (str) – The external user ID.

Returns:

The total number of personal records.

Return type:

int

get_all_UserPoints_by_gameId(gameId)

Retrieves all user points associated with a game ID.

Parameters:

gameId (int) – The game ID.

Returns:

A list of user points grouped by task and user.

Return type:

list

get_all_UserPoints_by_taskId(taskId)

Retrieves all user points for a specific task.

Parameters:

taskId (str) – The task ID.

Returns:

A list of user points.

Return type:

list

get_all_UserPoints_by_taskId_with_details(taskId)

Retrieves all user points for a specific task with details.

Parameters:

taskId (str) – The task ID.

Returns:

A list of user points with detailed information.

Return type:

list

get_avg_time_between_tasks_by_user_and_game_task(externalGameId, externalTaskId, externalUserId)

Retrieves the average time between tasks for a user and game task.

Parameters:

• externalGameId (str) – The external game ID.

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

Returns:

The average time between tasks.

Return type:

float

get_avg_time_between_tasks_for_all_users(externalGameId, externalTaskId)

Retrieves the average time between tasks for all users for a game task.

Parameters:

• externalGameId (str) – The external game ID.

• externalTaskId (str) – The external task ID.

Returns:

The average time between tasks for all users.

Return type:

float

get_first_user_points_in_external_task_id_by_user_id(externalTaskId, externalUserId)

Retrieves the first user points in an external task by user ID.

Parameters:

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

Returns:

The first user points in the task.

Return type:

UserPoints

get_global_avg_by_external_game_id(externalGameId)

Retrieves the global average time using data[“minutes”] for a given

external game ID.

Parameters:

externalGameId (str) – The external game ID.

Returns:

The global average time, or -1 if no valid data exists.

Return type:

float

get_global_calculation()

Calculates and retrieves a global performance metric.

Returns:

The calculated global performance metric.

Return type:

float

get_individual_calculation(userId)

Calculates and retrieves an individual performance metric for a user.

Parameters:

userId (str) – The user ID.

Returns:

The calculated individual performance metric.

Return type:

float

get_last_window_time_diff(externalTaskId, externalUserId)

Retrieves the time difference between the last two measurements by

a user for a task.

Parameters:

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

Returns:

The time difference between the last two measurements.

Return type:

float

get_new_last_window_time_diff(externalTaskId, externalUserId, externalGameId)

Retrieves the time difference between the last measurement and current

time for a user for a task in a game.

Parameters:

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

• externalGameId (str) – The external game ID.

Returns:

The time difference between the last measurement and

current time.

Return type:

float

get_personal_avg_by_external_game_id(externalGameId, externalUserId)

Retrieves the personal average time using data[“minutes”] for a given

external game ID and user ID.

Parameters:

• externalGameId (str) – The external game ID.

• externalUserId (str) – The external user ID.

Returns:

The personal average time, or -1 if no valid data exists.

Return type:

float

get_points_and_users_by_taskId(taskId)

Retrieves points and users associated with a task ID.

Parameters:

taskId (int) – The task ID.

Returns:

A list of user points with user information.

Return type:

list

get_start_time_for_last_task(userId)

Retrieves the start time of the last task completed by a user.

Parameters:

userId (str) – The user ID.

Returns:

The start time of the last task.

Return type:

datetime

get_task_and_sum_points_by_userId(userId)

Retrieves tasks and the sum of points for a user by their user ID.

Parameters:

userId (str) – The user ID.

Returns:

A list of tasks with the sum of points.

Return type:

list

get_task_by_externalUserId(externalUserId)

Retrieves tasks associated with a user by their external user ID.

Parameters:

externalUserId (str) – The external user ID.

Returns:

A list of tasks.

Return type:

list

get_time_taken_for_last_task(userId)

Retrieves the time taken by a user to complete the last task.

Parameters:

userId (str) – The user ID.

Returns:

The time taken to complete the last task.

Return type:

datetime

get_user_measurement_count(userId)

Retrieves the total number of measurements (tasks completed) by a

specific user.

Parameters:

userId (str) – The user ID.

Returns:

The total number of measurements completed by the user.

Return type:

int

get_user_task_measurements(externalTaskId, externalUserId)

Retrieves measurements for a user and task.

Parameters:

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

Returns:

A list of measurements for the user and task.

Return type:

list

get_user_task_measurements_count(externalTaskId, externalUserId)

Retrieves the total number of measurements by user and task.

Parameters:

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

Returns:

The total number of measurements by user and task.

Return type:

int

get_user_task_measurements_count_the_last_seconds(externalTaskId, externalUserId, seconds)

Retrieves the total number of measurements by user and task in the last

n seconds.

Parameters:

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

• seconds (int) – The number of seconds to consider.

Returns:

The total number of measurements by user and task in the last

n seconds.

Return type:

int

user_has_record_before_in_externalTaskId_last_min(externalTaskId, externalUserId, minutes)

Checks if a user has at least one record before in an external task.

Parameters:

• externalTaskId (str) – The external task ID.

• externalUserId (str) – The external user ID.

• minutes (int) – The number of minutes to consider.

Returns:

True if the user has one record before, False otherwise.

Return type:

bool

class app.repository.UserRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.users.Users'>)

Bases: BaseRepository

Repository class for users.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for users.

create_user_by_externalUserId(externalUserId: str)

Creates a new user with the provided external user ID.

Parameters:

externalUserId (str) – The external user ID.

Returns:

The created user.

Return type:

Users

class app.repository.WalletRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.wallet.Wallet'>)

Bases: BaseRepository

Repository class for wallets.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for wallets.

class app.repository.WalletTransactionRepository(session_factory: ~typing.Callable[[...], ~contextlib.AbstractContextManager[~sqlalchemy.orm.session.Session]], model=<class 'app.model.wallet_transactions.WalletTransactions'>)

Bases: BaseRepository

Repository class for wallet transactions.

session_factory

Factory for creating SQLAlchemy sessions.

Type:

Callable[…, AbstractContextManager[Session]]

model

SQLAlchemy model class for wallet transactions.