PaginatedQueryRunner

class safir.database.PaginatedQueryRunner(entry_type, cursor_type)

Bases: Generic[E, C]

Construct and run database queries that return paginated results.

This class implements the logic for keyset pagination based on arbitrary SQLAlchemy ORM where clauses.

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_count(session, stmt)

Count the number of objects that match a query.

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_count(session, stmt)

Count the number of objects that match a query.

There is nothing particular to pagination about this query, but it is often used in conjunction with pagination to provide the total count of matching entries, often in an X-Total-Count HTTP header.

Parameters:

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

  • stmt (Select) – Select statement to execute.

Returns:

Count of matching rows.

Return type:

int

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.

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.

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.

Return type:

PaginatedList

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.

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.

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.

Return type:

PaginatedList