app.repository.task_repository module

class app.repository.task_repository.TaskRepository(session_factory, model=<class 'app.model.tasks.Tasks'>, model_task_params=<class 'app.model.task_params.TasksParams'>, model_user_points=<class 'app.model.user_points.UserPoints'>)

Bases: BaseRepository

Repository class for tasks.

Parameters:

session_factory (Callable[[...], AbstractAsyncContextManager[AsyncSession]])

async read_by_gameId(schema, eager=False)

Reads tasks filtered by gameId (and any other schema fields).

Parameters:

eager (bool)

async read_by_gameId_and_externalTaskId(gameId, externalTaskId)

Look up a task by its game and external identifier.

Parameters:

• gameId – Internal identifier of the owning game.

• externalTaskId (str) – External identifier of the task.

Returns:

Tasks | None – The matching task, or None if not found.

async get_points_and_users_by_taskId(taskId)

Fetch a task by its internal id, raising if it does not exist.

Parameters:

taskId – Internal task identifier.

Returns:

Tasks – The matching task.

Raises:

NotFoundError – If no task has the given id.

async patch_by_id(taskId, fields)

Apply a small fields dict to the task identified by taskId. Returns the refreshed row. Used by the PATCH /games/{gameId}/tasks/{taskId} flow so the assignments admin view can rewrite strategyId (and status) without a full upsert.

Raises NotFoundError if the task does not exist.

Parameters:

fields (dict)

async delete_task_by_id(task_id)

Delete a single task and everything that hangs off it.

Mirrors the per-task branch of GameRepository.delete_game_by_id(): task params and the user-points rows that reference the task are removed first so the FK constraints don’t block the final DELETE on the task row. Returns True on success.

Raises NotFoundError if the task does not exist.

async list_by_strategy_id(strategy_id)

Return all tasks whose strategyId matches the given value.

Rollback cascade companion to GameRepository.list_by_strategy_id().

Parameters:

strategy_id (str)

async bulk_update_strategy_id(*, old_strategy_id, new_strategy_id)

Rewrite every task’s strategyId from old_strategy_id to new_strategy_id in a single UPDATE. Returns the row count.

Rollback cascade companion to GameRepository.bulk_update_strategy_id().

Parameters:

• old_strategy_id (str)

• new_strategy_id (str)

Return type:

int