app.repository.game_repository module

class app.repository.game_repository.GameRepository(session_factory, 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.

Parameters:

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

async get_all_games(schema, api_key=None, oauth_user_id=None, is_admin=False)

List games visible to the caller, filtered, ordered and paginated.

Non-null fields on schema become filters and drive ordering/pagination. Unless is_admin is True, results are restricted to games owned by the supplied api_key / oauth_user_id.

Parameters:

• schema – Search schema with optional filters and ordering/page/page_size.

• api_key – API key of the caller, used for ownership scoping.

• oauth_user_id – OAuth subject of the caller, used for scoping.

• is_admin (bool) – When True, bypass ownership scoping and list all games.

Returns:

A paginated result with the matching games and search metadata.

async get_game_by_id(id)

Fetch a game together with its configured parameters.

Parameters:

id (str) – Internal game identifier.

Returns:

BaseGameResult – The game and its params (id/key/value).

Raises:

NotFoundError – If no game has the given id.

async patch_game_by_id(gameId, schema)

Partially update a game’s columns from schema.

Only non-null fields on schema are written to the row.

Parameters:

• gameId (str) – Internal game identifier.

• schema – Patch schema whose non-null fields are applied.

Returns:

The updated game.

Raises:

NotFoundError – If no game has the given id.

async list_by_strategy_id(strategy_id)

Return all games whose strategyId matches the given value.

Used by the rollback cascade to know which games will be reassigned to the rolled-back version (the actual UPDATE goes through bulk_update_strategy_id(); this helper is exposed for audit logging and tests).

Parameters:

strategy_id (str)

async list_external_ids(ids)

Map internal game id → externalGameId for the given ids.

Used by the strategy-usage view to render the parent game of a task-level assignment by its human-readable external id instead of a raw UUID, in a single query rather than N reads.

Return type:

dict

async bulk_update_strategy_id(*, old_strategy_id, new_strategy_id)

Rewrite every game’s strategyId from old_strategy_id to new_strategy_id in a single UPDATE. Returns the row count so the caller can log/audit the cascade.

Used by the rollback flow: when a published custom strategy is rolled back, the games that pointed at the previous UUID get reassigned to the target UUID in one trip so no game is left referring to an ARCHIVED row.

Parameters:

• old_strategy_id (str)

• new_strategy_id (str)

Return type:

int

async delete_game_by_id(game_id)

Delete a game by its internal identifier.

Parameters:

game_id (str) – Internal game identifier.

Raises:

NotFoundError – If no game has the given id.