CountedPaginatedQueryRunner

class safir.database.CountedPaginatedQueryRunner(entry_type, cursor_type)

Bases: PaginatedQueryRunner[E, C]

Run database queries that return paginated results with counts.

This variation of PaginatedQueryRunner always runs a second query to count the total number of available entries if queried without pagination. It should only be used on small tables or with queries that can be satisfied from the table indices; otherwise, the count query could be undesirably slow.

Parameters:

• entry_type (type[TypeVar(E, bound= BaseModel)]) – Type of each entry returned by the queries. This must be a Pydantic model.

• cursor_type (type[TypeVar(C, bound= PaginationCursor)]) – Type of the pagination cursor, which encapsulates the logic of how entries are sorted and what set of keys is used to retrieve the next or previous batch of entries.

Methods Summary

query_object(session, stmt, *[, cursor, limit])	Perform a query for objects with an optional cursor and limit.
query_row(session, stmt, *[, cursor, limit])	Perform a query for attributes with an optional cursor and limit.

Methods Documentation

async query_object(session, stmt, *, cursor=None, limit=None)

Perform a query for objects with an optional cursor and limit.

Perform the query provided in stmt with appropriate sorting and pagination as determined by the cursor type. Also performs a second query to get the total count of entries if retrieved without pagination.

This method should be used with queries that return a single SQLAlchemy model. The provided query will be run with the session scalars method and the resulting object passed to Pydantic’s model_validate to convert to entry_type. For queries returning a tuple of attributes, use query_row instead.

Unfortunately, this distinction cannot be type-checked, so be careful to use the correct method.

Parameters:

• session (async_scoped_session) – Database session within which to run the query.

• stmt (Select) – Select statement to execute. Pagination and ordering will be added, so this statement should not already have limits or order clauses applied. This statement must return a list of SQLAlchemy ORM models that can be converted to entry_type by Pydantic.

• cursor (Optional[TypeVar(C, bound= PaginationCursor)], default: None) – If present, continue from the provided keyset cursor.

• limit (int | None, default: None) – If present, limit the result count to at most this number of rows.

Returns:

Results of the query wrapped with pagination information and a count of the total number of entries.

Return type:

CountedPaginatedList

async query_row(session, stmt, *, cursor=None, limit=None)

Perform a query for attributes with an optional cursor and limit.

Perform the query provided in stmt with appropriate sorting and pagination as determined by the cursor type. Also performs a second query to get the total count of entries if retrieved without pagination.

This method should be used with queries that return a list of attributes that can be converted to the entry_type Pydantic model. For queries returning a single ORM object, use query_object instead.

Unfortunately, this distinction cannot be type-checked, so be careful to use the correct method.

Parameters:

• session (async_scoped_session) – Database session within which to run the query.

• stmt (Select) – Select statement to execute. Pagination and ordering will be added, so this statement should not already have limits or order clauses applied. This statement must return a tuple of attributes that can be converted to entry_type by Pydantic’s model_validate.

• cursor (Optional[TypeVar(C, bound= PaginationCursor)], default: None) – If present, continue from the provided keyset cursor.

• limit (int | None, default: None) – If present, limit the result count to at most this number of rows.

Returns:

Results of the query wrapped with pagination information and a count of the total number of entries.

Return type:

CountedPaginatedList