pypaginate.domain.protocols

Backend contracts (ports) for the hexagonal architecture.

Each protocol defines the minimal interface a backend must implement. Protocols are generic over query type Q and item type T for full type safety in backend implementations.

Classes

CursorBackend

Async backend for cursor/keyset-based pagination.

FilterBackend

Translates filter specs to backend query conditions.

PaginationBackend

Async backend for offset-based pagination.

SearchBackend

Translates search specs to backend query conditions.

SortBackend

Translates sort specs to backend query ordering.

SyncPaginationBackend

Sync backend for offset-based pagination.

Module Contents

class pypaginate.domain.protocols.CursorBackend

Bases: Protocol[T]

Async backend for cursor/keyset-based pagination.

async fetch_page(query: object, *, limit: int, after: str | None = None, before: str | None = None) tuple[list[T], str | None, str | None]

Fetch a page: returns (items, next_cursor, prev_cursor).

class pypaginate.domain.protocols.FilterBackend

Bases: Protocol

Translates filter specs to backend query conditions.

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

Apply filter specifications to a query.

class pypaginate.domain.protocols.PaginationBackend

Bases: Protocol[T]

Async backend for offset-based pagination.

async count(query: object) int

Count total items matching the query.

async fetch(query: object, offset: int, limit: int) list[T]

Fetch a slice of items from the query.

class pypaginate.domain.protocols.SearchBackend

Bases: Protocol

Translates search specs to backend query conditions.

Apply search specification to a query.

class pypaginate.domain.protocols.SortBackend

Bases: Protocol

Translates sort specs to backend query ordering.

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

Apply sort specifications to a query.

class pypaginate.domain.protocols.SyncPaginationBackend

Bases: Protocol[T]

Sync backend for offset-based pagination.

count(query: object) int

Count total items matching the query.

fetch(query: object, offset: int, limit: int) list[T]

Fetch a slice of items from the query.