pypaginate.adapters.sqlalchemy

SQLAlchemy backends for pagination, filtering, sorting, and search.

Submodules

Classes

`SQLAlchemyBackend`	Async offset pagination backend for SQLAlchemy.
`SQLAlchemyCursorBackend`	Async cursor/keyset pagination backend.
`SQLAlchemyFilterBackend`	Translates FilterSpec to SQLAlchemy WHERE clauses.
`SQLAlchemySearchBackend`	Translates SearchSpec to SQLAlchemy ILIKE clauses.
`SQLAlchemySortBackend`	Translates SortSpec to SQLAlchemy ORDER BY clauses.
`SyncSQLAlchemyBackend`	Sync offset pagination backend for SQLAlchemy.
`SyncSQLAlchemyCursorBackend`	Sync cursor/keyset pagination backend.

Package Contents

class pypaginate.adapters.sqlalchemy.SQLAlchemyBackend(session: sqlalchemy.ext.asyncio.AsyncSession, *, count_query: object | None = None, unique: bool = False)

Bases: Generic[ItemT]

Async offset pagination backend for SQLAlchemy.

Satisfies PaginationBackend[ItemT] protocol.

Parameters:: session – An async SQLAlchemy session.

async count(query: object) → int: Count rows. Uses custom count query if provided.

async fetch(query: object, offset: int, limit: int) → list[ItemT]: Fetch rows with OFFSET/LIMIT. Deduplicates if unique=True.

class pypaginate.adapters.sqlalchemy.SQLAlchemyCursorBackend(session: sqlalchemy.ext.asyncio.AsyncSession)

Bases: Generic[ItemT]

Async cursor/keyset pagination backend.

Satisfies CursorBackend[ItemT] protocol.

Parameters:: session – An async SQLAlchemy session.

async fetch_page(query: sqlalchemy.sql.Select[Any], *, limit: int, after: str | None = None, before: str | None = None) → tuple[list[ItemT], str | None, str | None]

Fetch a keyset-paginated page.

Parameters:

query – A SQLAlchemy Select with ORDER BY.
limit – Maximum items per page.
after – Cursor for the next page.
before – Cursor for the previous page.

Returns:

Tuple of (items, next_cursor, prev_cursor).

class pypaginate.adapters.sqlalchemy.SQLAlchemyFilterBackend

Translates FilterSpec to SQLAlchemy WHERE clauses.

Satisfies FilterBackend protocol.

apply_filters(query: object, filters: collections.abc.Sequence[pypaginate.domain.specs.FilterSpec]) → object

Apply filter specs to a SQLAlchemy Select.

Parameters:

query – A SQLAlchemy Select statement.
filters – Filter specifications to apply.

Returns:

Modified Select with WHERE clauses.

class pypaginate.adapters.sqlalchemy.SQLAlchemySearchBackend

Translates SearchSpec to SQLAlchemy ILIKE clauses.

Satisfies SearchBackend protocol. Tokenizes the query and matches each token against all specified fields.

static apply_search(query: object, spec: pypaginate.domain.specs.SearchSpec) → object

Apply a search spec to a SQLAlchemy Select.

Parameters:

query – A SQLAlchemy Select statement.
spec – Search specification with query and fields.

Returns:

Modified Select with WHERE clauses for search.

class pypaginate.adapters.sqlalchemy.SQLAlchemySortBackend

Translates SortSpec to SQLAlchemy ORDER BY clauses.

Satisfies SortBackend protocol.

static apply_sorting(query: object, sorting: collections.abc.Sequence[pypaginate.domain.specs.SortSpec]) → object

Apply sort specs to a SQLAlchemy Select.

Parameters:

query – A SQLAlchemy Select statement.
sorting – Sort specifications (applied in order).

Returns:

Modified Select with ORDER BY clauses.

class pypaginate.adapters.sqlalchemy.SyncSQLAlchemyBackend(session: sqlalchemy.orm.Session, *, count_query: object | None = None, unique: bool = False)

Bases: Generic[ItemT]

Sync offset pagination backend for SQLAlchemy.

Satisfies SyncPaginationBackend[ItemT] protocol.

Parameters:: session – A synchronous SQLAlchemy session.

count(query: object) → int: Count rows. Uses custom count query if provided.

fetch(query: object, offset: int, limit: int) → list[ItemT]: Fetch rows with OFFSET/LIMIT. Deduplicates if unique=True.

class pypaginate.adapters.sqlalchemy.SyncSQLAlchemyCursorBackend(session: sqlalchemy.orm.Session)

Bases: Generic[ItemT]

Sync cursor/keyset pagination backend.

Satisfies cursor backend contract for synchronous sessions.

Parameters:: session – A synchronous SQLAlchemy session.

fetch_page(query: sqlalchemy.sql.Select[Any], *, limit: int, after: str | None = None, before: str | None = None) → tuple[list[ItemT], str | None, str | None]

Fetch a keyset-paginated page.

Parameters:

query – A SQLAlchemy Select with ORDER BY.
limit – Maximum items per page.
after – Cursor for the next page.
before – Cursor for the previous page.

Returns:

Tuple of (items, next_cursor, prev_cursor).