pypaginate.adapters.sqlalchemy.backend

Offset pagination backends for SQLAlchemy (async and sync).

Implements PaginationBackend[T] and SyncPaginationBackend[T] protocols using SELECT COUNT(*) for counting and OFFSET/LIMIT for fetching.

Classes

SQLAlchemyBackend

Async offset pagination backend for SQLAlchemy.

SyncSQLAlchemyBackend

Sync offset pagination backend for SQLAlchemy.

Module Contents

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