app.services.user_points.queries module

Read-only points queries.

Aggregations and lookups over user_points for games, tasks and users. These methods never mutate state; the write path lives in app.services.user_points.assignment.

class app.services.user_points.queries.PointsQueryMixin[source]

Bases: UserPointsContext

async query_user_points(schema)[source]

Run a filtered, paginated query over the user_points table.

Parameters:

schema – Search schema with filters and ordering/pagination.

Returns:

Any – Items plus search metadata, as returned by the repository.

Return type:

Any

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

List a game’s tasks and, per task, the users who earned points.

For each user the response includes their first action timestamp on that task.

Parameters:

  • gameId – Internal game identifier.

  • api_key (str) – Caller’s API key, used when enforce_scope.

  • oauth_user_id (str) – Caller’s OAuth subject, used when scoping.

  • is_admin (bool) – Whether the caller has the admin role.

  • enforce_scope (bool) – When True, verify the caller may access the game before returning data.

Returns:

ListTasksWithUsers – Tasks each paired with their participating users.

Raises:

NotFoundError – If the game or its tasks do not exist.

Return type:

ListTasksWithUsers

async get_points_by_user_list(users_list, *, api_key=None, oauth_user_id=None, is_admin=False, enforce_scope=False)[source]

Fetch per-game point totals for a list of users, concurrently.

Fans out one get_all_points_by_externalUserId() call per user, bounded by a concurrency semaphore.

Parameters:

  • users_list – External user identifiers to look up.

  • api_key (str) – Caller’s API key, used when enforce_scope.

  • oauth_user_id (str) – Caller’s OAuth subject, used when scoping.

  • is_admin (bool) – Whether the caller has the admin role.

  • enforce_scope (bool) – When True, enforce per-user access checks.

Returns:

list[UserGamePoints] – One aggregate result per requested user.

Return type:

list[UserGamePoints]

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

Return a user’s points across every game they participate in.

Resolves the user, finds the games behind their points rows, and aggregates each game’s detailed points concurrently.

Parameters:

  • externalUserId – External identifier of the user.

  • api_key (str) – Caller’s API key, used when enforce_scope.

  • oauth_user_id (str) – Caller’s OAuth subject, used when scoping.

  • is_admin (bool) – Whether the caller has the admin role.

  • enforce_scope (bool) – When True, verify access to the user.

Returns:

list[AllPointsByGame] – Detailed points grouped per game.

Raises:

NotFoundError – If the user does not exist.

Return type:

list[AllPointsByGame]

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

Aggregate all points awarded within a game.

Parameters:

  • gameId – Internal game identifier.

  • api_key (str) – Caller’s API key, used when enforce_scope.

  • oauth_user_id (str) – Caller’s OAuth subject, used when scoping.

  • is_admin (bool) – Whether the caller has the admin role.

  • enforce_scope (bool) – When True, verify access to the game.

Returns:

AllPointsByGame – The game’s aggregated points.

Raises:

NotFoundError – If the game does not exist.

Return type:

AllPointsByGame

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

Aggregate a game’s points including per-award detail.

Like get_points_by_gameId() but the result carries the detailed per-award breakdown for each task/user.

Parameters:

  • gameId (UUID) – Internal game identifier.

  • api_key (str) – Caller’s API key, used when enforce_scope.

  • oauth_user_id (str) – Caller’s OAuth subject, used when scoping.

  • is_admin (bool) – Whether the caller has the admin role.

  • enforce_scope (bool) – When True, verify access to the game.

Returns:

AllPointsByGame – The game’s aggregated points with detail.

Raises:

NotFoundError – If the game does not exist.

Return type:

AllPointsByGame

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

Return one user’s point awards within a single game.

Parameters:

  • gameId – Internal game identifier.

  • externalUserId – External identifier of the user.

  • api_key (str) – Caller’s API key, used when enforce_scope.

  • oauth_user_id (str) – Caller’s OAuth subject, used when scoping.

  • is_admin (bool) – Whether the caller has the admin role.

  • enforce_scope (bool) – When True, verify access to the game.

Returns:

list[PointsAssignedToUser] – The user’s awards in the game.

Raises:

NotFoundError – If the game or user does not exist.

Return type:

list[PointsAssignedToUser]

async get_users_points_by_externalGameId(externalGameId)[source]

Return per-user point totals for a game identified by external id.

Parameters:

externalGameId – External identifier of the game.

Returns:

list[ResponseGetPointsByGame] – Aggregated points per user/task in the game.

Raises:

NotFoundError – If the game or its tasks do not exist.

Return type:

list[ResponseGetPointsByGame]

async get_users_points_by_externalTaskId(externalTaskId)[source]

Return per-user point totals for a task identified by external id.

Parameters:

externalTaskId – External identifier of the task.

Returns:

list[ResponseGetPointsByTask] – Aggregated points per user on the task.

Raises:

NotFoundError – If the task does not exist.

Return type:

list[ResponseGetPointsByTask]

async get_users_points_by_externalTaskId_and_externalUserId(externalTaskId, externalUserId)[source]

Return a single user’s points on a single task (both by external id).

Parameters:

  • externalTaskId – External identifier of the task.

  • externalUserId – External identifier of the user.

Returns:

Any – The user’s points for that task.

Raises:

NotFoundError – If the task or user does not exist.

Return type:

Any

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

Return one user’s points aggregated across all their games.

Parameters:

  • externalUserId – External identifier of the user.

  • api_key (str) – Caller’s API key, used when enforce_scope.

  • oauth_user_id (str) – Caller’s OAuth subject, used when scoping.

  • is_admin (bool) – Whether the caller has the admin role.

  • enforce_scope (bool) – When True, verify access to the user.

Returns:

UserGamePoints – The user’s points grouped by game.

Raises:

NotFoundError – If the user does not exist.

Return type:

UserGamePoints

async get_points_of_user(externalUserId)[source]

Return a user’s total points plus a per-task breakdown.

Parameters:

externalUserId – External identifier of the user.

Returns:

ResponsePointsByExternalUserId – The summed total and the per-task points list.

Raises:

NotFoundError – If the user does not exist.

Return type:

ResponsePointsByExternalUserId

async get_points_of_simulated_task(externalTaskId, simulationHash)[source]

Return points rows produced by a specific simulation run of a task.

Parameters:

  • externalTaskId – External identifier of the task.

  • simulationHash – Hash identifying the simulation run.

Returns:

Any – The points rows belonging to that simulation.

Return type:

Any

async get_all_point_of_tasks_list(list_ids_tasks, withData=False)[source]

Return all points rows for a list of task ids.

Parameters:

  • list_ids_tasks – Internal task ids to fetch points for.

  • withData (bool) – When True, include the full JSON data column; otherwise return a lighter projection.

Returns:

Any – The matching points rows.

Return type:

Any