app.services.task_service module

class app.services.task_service.TaskService(strategy_service, task_repository, game_repository, user_repository, user_points_repository, game_params_repository, task_params_repository, strategy_definition_service=None)[source]

Bases: BaseService

Service class for managing tasks.

Parameters:
strategy_service

Service instance for strategies.

Type:

StrategyService

task_repository

Repository instance for tasks.

Type:

TaskRepository

game_repository

Repository instance for games.

Type:

GameRepository

user_repository

Repository instance for users.

Type:

UserRepository

user_points_repository

Repository instance for user points.

Type:

UserPointsRepository

game_params_repository

Repository instance for game parameters.

Type:

GameParamsRepository

task_params_repository

Repository instance for task parameters.

Type:

TaskParamsRepository

__init__(strategy_service, task_repository, game_repository, user_repository, user_points_repository, game_params_repository, task_params_repository, strategy_definition_service=None)[source]
Initializes the TaskService with the provided repositories and

services.

strategy_definition_service is optional so existing call sites and tests that don’t exercise custom: strategyIds keep working. Required for patch_task_by_id() to validate custom: ids against the persistent registry.

Parameters:
Return type:

None

async patch_task_by_id(gameId, taskId, schema, *, api_key=None, oauth_user_id=None, is_admin=False, enforce_scope=False)[source]

Partially update a task identified by gameId + taskId.

Mutable fields: strategyId, status and params. When params is provided it is treated as the desired full set and synced via _sync_task_params() (update existing rows by id, create rows without an id, delete rows omitted from the list); omit params to leave them untouched.

strategyId accepts both built-ins and custom:<uuid> - validated by _validate_strategy_assignment() with the caller’s tenant boundary.

Parameters:
  • gameId (UUID)

  • taskId (UUID)

  • schema (PatchTask)

  • api_key (str | None)

  • oauth_user_id (str | None)

  • is_admin (bool)

  • enforce_scope (bool)

Return type:

ResponsePatchTask

async delete_task_by_id(gameId, taskId, *, api_key=None, oauth_user_id=None, is_admin=False, enforce_scope=False)[source]

Delete a task (and its params/points) identified by gameId + taskId.

Full-stack addition mirroring GameService.delete_game_by_id(): the cascade lives in TaskRepository.delete_task_by_id() so task params and the user-points rows that reference the task go before the task row.

Parameters:
  • gameId (UUID)

  • taskId (UUID)

  • api_key (str | None)

  • oauth_user_id (str | None)

  • is_admin (bool)

  • enforce_scope (bool)

Return type:

ResponseDeleteTask

async duplicate_task(gameId, taskId, externalTaskId, *, api_key=None, oauth_user_id=None, is_admin=False, enforce_scope=False)[source]

Duplicate a task within the same game under a new externalTaskId.

Deep-copies the source task’s strategy and params onto the new task. Reuses create_task_by_game_id() for the heavy lifting (uniqueness check on the new externalTaskId, row + param insertion, response shaping) so the duplicate path can’t drift from normal task creation.

Parameters:
  • gameId (UUID)

  • taskId (UUID)

  • externalTaskId (str)

  • api_key (str | None)

  • oauth_user_id (str | None)

  • is_admin (bool)

  • enforce_scope (bool)

Return type:

CreateTaskPostSuccesfullyCreated

async get_tasks_list_by_externalGameId(externalGameId, find_query)[source]
Retrieves a list of tasks associated with a game by its external game

ID.

Parameters:
  • externalGameId (str) – The external game ID.

  • find_query – The query for finding tasks.

Returns:

list – A list of tasks associated with the game.

Return type:

dict[str, Any]

async get_tasks_list_by_gameId(gameId, find_query, *, api_key=None, oauth_user_id=None, is_admin=False, enforce_scope=False)[source]

Retrieves a list of tasks associated with a game by its game ID.

Parameters:
  • gameId (UUID) – The game ID.

  • find_query – The query for finding tasks.

  • api_key (str)

  • oauth_user_id (str)

  • is_admin (bool)

  • enforce_scope (bool)

Returns:

list – A list of tasks associated with the game.

Return type:

dict[str, Any]

async get_task_by_gameId_externalTaskId(gameId, externalTaskId, *, api_key=None, oauth_user_id=None, is_admin=False, enforce_scope=False)[source]

Retrieves a task by its game ID and external task ID.

Parameters:
  • gameId (UUID) – The game ID.

  • externalTaskId (str) – The external task ID.

  • api_key (str)

  • oauth_user_id (str)

  • is_admin (bool)

  • enforce_scope (bool)

Returns:

CreateTaskPostSuccesfullyCreated – The task details.

Return type:

CreateTaskPostSuccesfullyCreated

async get_task_by_externalGameId_externalTaskId(gameId, externalTaskId, *, api_key=None, oauth_user_id=None, is_admin=False, enforce_scope=False)[source]

Retrieves a task by its game ID and external task ID.

Parameters:
  • gameId (UUID) – The game ID.

  • externalTaskId (str) – The external task ID.

  • api_key (str)

  • oauth_user_id (str)

  • is_admin (bool)

  • enforce_scope (bool)

Returns:

CreateTaskPostSuccesfullyCreated – The task details.

Return type:

CreateTaskPostSuccesfullyCreated

async create_task_by_externalGameId(externalGameId, create_query)[source]

Creates a task for a game by its external game ID.

Parameters:
  • externalGameId (str) – The external game ID.

  • create_query – The query for creating the task.

Returns:

CreateTaskPostSuccesfullyCreated – The created task details.

Return type:

CreateTaskPostSuccesfullyCreated

async create_task_by_game_id(gameId, create_query, api_key=None, *, oauth_user_id=None, is_admin=False, enforce_scope=False)[source]

Creates a task for a game by its game ID.

Parameters:
  • gameId (UUID) – The game ID.

  • create_query – The query for creating the task.

  • api_key (str)

  • oauth_user_id (str)

  • is_admin (bool)

  • enforce_scope (bool)

Returns:

CreateTaskPostSuccesfullyCreated – The created task details.

Return type:

CreateTaskPostSuccesfullyCreated

async get_task_detail_by_id(schema)[source]

Retrieves task details by its ID.

Parameters:

schema – The schema containing the task ID.

Returns:

dict – The task and strategy details.

Return type:

dict[str, Any]

async get_points_by_task_id(gameId, externalTaskId, *, api_key=None, oauth_user_id=None, is_admin=False, enforce_scope=False)[source]

Retrieves points by task ID.

Parameters:
  • gameId (UUID) – The game ID.

  • externalTaskId (str) – The external task ID.

  • api_key (str)

  • oauth_user_id (str)

  • is_admin (bool)

  • enforce_scope (bool)

Returns:

list – A list of points associated with the task.

Return type:

Any

async get_points_of_user_by_task_id(gameId, externalTaskId, externalUserId, *, api_key=None, oauth_user_id=None, is_admin=False, enforce_scope=False)[source]

Retrieves points of a user by task ID.

Parameters:
  • gameId (UUID) – The game ID.

  • externalTaskId (str) – The external task ID.

  • externalUserId (str) – The external user ID.

  • api_key (str)

  • oauth_user_id (str)

  • is_admin (bool)

  • enforce_scope (bool)

Returns:

dict – The user’s points details.

Return type:

Any

async get_points_by_task_id_with_details(gameId, externalTaskId, *, api_key=None, oauth_user_id=None, is_admin=False, enforce_scope=False)[source]

Retrieves points by task ID with details.

Parameters:
  • gameId (UUID) – The game ID.

  • externalTaskId (str) – The external task ID.

  • api_key (str)

  • oauth_user_id (str)

  • is_admin (bool)

  • enforce_scope (bool)

Returns:

list – A list of points associated with the task.

Return type:

Any

async get_task_params_by_externalTaskId(externalTaskId)[source]

Retrieves task parameters by external task ID.

Parameters:

externalTaskId (str) – The external task ID.

Returns:

list – A list of task parameters associated with the task.

Return type:

Any