pypaginate.adapters.sqlalchemy.cursor

Cursor/keyset pagination backends (async and sync).

Implements CursorBackend[T] protocol using built-in cursor encoding and keyset WHERE clause construction. Requires the query to have an ORDER BY clause.

Classes

SQLAlchemyCursorBackend	Async cursor/keyset pagination backend.
SyncSQLAlchemyCursorBackend	Sync cursor/keyset pagination backend.

Module Contents

class pypaginate.adapters.sqlalchemy.cursor.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.cursor.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).