Pagination¶
The seven strategies, with their parameters and pages.
Each strategy in its module of fastapi_alcazaba.pagination, with its parameters and page.
In the guide: Pagination. In an app: the examples.
fastapi_alcazaba.pagination.page_number.PaginationPageNumber
¶
PaginationPageNumber(*, default_page_size: int = 50, max_page_size: int = 100)
Pagination by page number: ?page=2&page_size=20.
pages = PaginationPageNumber()
@app.get("/products")
@cache(expire=60)
async def products(
params: Annotated[PageNumberParams, Depends(pages)],
) -> PageNumberPage[Product]:
return pages.paginate(rows, params, total=count)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
default_page_size
|
int
|
items per page when the request doesn't say. |
50
|
max_page_size
|
int
|
the most a request may ask for (a 422 beyond), so that nobody asks for a million rows at once. |
100
|
params
¶
params(*, page: int = 1, page_size: int | None = None) -> PageNumberParams
The parameters of a page, as the dependency gives them (for tests and scripts).
paginate
¶
paginate(items: Sequence[T], params: PageNumberParams, *, total: int | None = None) -> PageNumberPage[T]
The page of items that params asks for. Without total, items is the whole
list and the page is sliced from it; with total, items is that page already (queried
with params.offset and params.limit). The response links to the first, previous,
next and last pages.
fastapi_alcazaba.pagination.offset.PaginationOffset
¶
PaginationOffset(*, default_limit: int = 50, max_limit: int = 100)
Pagination by offset: ?offset=40&limit=20.
offsets = PaginationOffset()
@app.get("/products")
@cache(expire=60)
async def products(params: Annotated[OffsetParams, Depends(offsets)]) -> OffsetPage[Item]:
rows = ... # ORDER BY id LIMIT params.limit OFFSET params.offset
return offsets.paginate(rows, params, total=...) # SELECT count(*)
With total, the page has it and links to the last page too. Without it, query
params.fetch rows (one more than the page) and pass queried=True: the extra row says
whether there's a next page, without a COUNT(*), slow on big tables and outdated as soon as
rows are written.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
default_limit
|
int
|
items per page when the request doesn't say. |
50
|
max_limit
|
int
|
the most a request may ask for (a 422 beyond). |
100
|
params
¶
params(*, offset: int = 0, limit: int | None = None) -> OffsetParams
The parameters of a page, as the dependency gives them (for tests and scripts).
paginate
¶
paginate(items: Sequence[T], params: OffsetParams, *, total: int | None = None, queried: bool = False) -> OffsetPage[T]
The page of items that params asks for, linked to the first, previous and next
pages, and to the last one when the total is known.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
total
|
int | None
|
the count of the whole list: |
None
|
queried
|
bool
|
without a total, With neither, |
False
|
fastapi_alcazaba.pagination.ranges.PaginationRange
¶
PaginationRange(*, default_items: int = 50, max_items: int = 100)
Pagination by range: ?start=0&end=24, both included.
ranges = PaginationRange()
@app.get("/contacts")
@cache(expire=60)
async def contacts(params: Annotated[RangeParams, Depends(ranges)]) -> RangePage[Contact]:
count = ... # SELECT count(*)
rows = ... # ORDER BY name LIMIT params.limit OFFSET params.offset
return ranges.paginate(rows, params, total=count)
The response says which items it holds in Content-Range: items 0-24/319 (items */319
when it holds none), on top of the page's start, end and total. A range that ends
before it starts, or holds more than max_items items, is a 400; one that starts past the
last item a 416, with Content-Range: items */<total>.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
default_items
|
int
|
items in the range when the request gives no |
50
|
max_items
|
int
|
the most a range may hold (a 400 beyond). |
100
|
params
¶
params(*, start: int = 0, end: int | None = None) -> RangeParams
The parameters of a range, as the dependency gives them (for tests and scripts).
Raises an HTTPException (400) if end is before start, or the range holds more than
max_items items.
paginate
¶
paginate(items: Sequence[T], params: RangeParams, *, total: int | None = None) -> RangePage[T]
The range of items that params asks for. Without total, items is the whole
list and the range is sliced from it; with total, items is that range already
(queried with params.offset and params.limit). The response links to the first,
previous, next and last ranges of the same size. Raises an HTTPException (416) when the
range starts past the last item of a list that isn't empty.
fastapi_alcazaba.pagination.cursor.PaginationCursor
¶
PaginationCursor(secret: Secrets, *, ttl: int | None = None, default_limit: int = 50, max_limit: int = 100)
Pagination by cursor, in both directions: ?limit=20, then ?cursor=<next_cursor> for
the next page or ?cursor=<prev_cursor> for the one before.
cursors = PaginationCursor(settings.cursor_secret)
@app.get("/events")
@cache(expire=60)
async def events(params: Annotated[CursorParams, Depends(cursors)]) -> CursorPage[Event]:
rows = ... # WHERE (created_at, id) > params.after ORDER BY ... LIMIT params.fetch,
# or WHERE (created_at, id) < params.before ORDER BY ... DESC LIMIT params.fetch
return cursors.paginate(rows, params, key=lambda e: (e.created_at, e.id), queried=True)
A cursor holds the sort key of the first or last item of a page, a digest of the request's
other query parameters and when it was issued, signed with HMAC-SHA256. A cursor that a
client made or changed, that comes with other query parameters than the ones it was issued
for (the next page of genre=poetry asked with genre=novel), or that is older than ttl,
is a 400 before the endpoint runs. limit may change from one page to the next. Cursors are
base64, not encrypted: a client can read the sort key in them, so sort by what it may see.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
secret
|
Secrets
|
at least 32 bytes ( |
required |
ttl
|
int | None
|
seconds a cursor is accepted for, from when its page was made; None for ever. |
None
|
default_limit
|
int
|
items per page when the request doesn't say. |
50
|
max_limit
|
int
|
the most a request may ask for (a 422 beyond). |
100
|
params
¶
params(*, cursor: str | None = None, limit: int | None = None, query: Sequence[tuple[str, str]] = ()) -> CursorParams
The parameters of a page, with the cursor checked and decoded, as the dependency gives them.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
Sequence[tuple[str, str]]
|
the request's query parameters, which the cursor must have been issued
for; Raises an HTTPException (400) if the cursor isn't valid, has expired, or was issued for other query parameters. |
()
|
encode
¶
encode(position: Position, *, before: bool = False, query: Sequence[tuple[str, str]] = ()) -> str
The cursor of the page after position (before it, with before), for requests with
these other query parameters: for links of your own, such as the page that starts at a
date.
paginate
¶
paginate(items: Sequence[T], params: CursorParams, *, key: Callable[[T], Position], descending: bool = False, queried: bool = False) -> CursorPage[T]
The page of items that params asks for, in the list's order whichever the
direction, with the cursors of the pages around it, bound to the same query parameters.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
Callable[[T], Position]
|
an item's sort key: a unique value or tuple ( |
required |
descending
|
bool
|
the list goes from the greatest key to the smallest (newest first). |
False
|
queried
|
bool
|
|
False
|
fastapi_alcazaba.pagination.keyset.PaginationKeyset
¶
PaginationKeyset(sort: type[CursorValue] | Sequence[type[CursorValue]], *, default_limit: int = 50, max_limit: int = 100)
Pagination by keyset, with the keys in plain sight: ?limit=20, then
?after=<after>&limit=20.
messages = PaginationKeyset(sort=int) # by id: ?after=42
@app.get("/messages")
@cache(expire=60)
async def list_messages(
params: Annotated[KeysetParams, Depends(messages)],
) -> KeysetPage[Message]:
rows = ... # WHERE id > params.after ORDER BY id LIMIT params.fetch
return messages.paginate(rows, params, key=lambda m: m.id, queried=True)
events = PaginationKeyset(sort=(datetime, int)) # ?after=["2026-10-29T10:00:00Z",42]
The page gives the last item's key as after, and a client can as well start after a key of
its own (the events after a date). Keys are neither encoded nor signed, so sort by what
clients may see. Each value is converted to the type sort declares, and a value of another
type, another number of them or JSON that doesn't parse is a 400, before the endpoint runs.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sort
|
type[CursorValue] | Sequence[type[CursorValue]]
|
the type of the sort key ( |
required |
default_limit
|
int
|
items per page when the request doesn't say. |
50
|
max_limit
|
int
|
the most a request may ask for (a 422 beyond). |
100
|
params
¶
params(*, after: str | None = None, limit: int | None = None) -> KeysetParams
The parameters of a page, with after converted, as the dependency gives them.
Raises an HTTPException (400) if it isn't valid.
decode
¶
decode(after: str) -> Position
The sort key in after, converted to the types of sort: a value as text, or a JSON
array of them. Raises an HTTPException (400) if it isn't that.
encode
¶
encode(position: Position) -> str
position as after holds it. Raises TypeError if it isn't of the types of sort.
paginate
¶
paginate(items: Sequence[T], params: KeysetParams, *, key: Callable[[T], Position], descending: bool = False, queried: bool = False) -> KeysetPage[T]
The page of items that params asks for.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
Callable[[T], Position]
|
an item's sort key, of the types of |
required |
descending
|
bool
|
|
False
|
queried
|
bool
|
|
False
|
fastapi_alcazaba.pagination.scroll.PaginationScroll
¶
PaginationScroll(item_type: type[T], secret: Secrets, *, default_batch_size: int = 100, max_batch_size: int = 1000, keep_alive: int = 60, max_items: int = 10000, cache: Cache | None = None, namespace: str = 'scroll')
Bases: Generic[T]
Pagination by scroll: ?batch_size=100 keeps what the query found as a snapshot in the
cache's backend and answers its first batch, then ?scroll_id=<scroll_id> the next one.
scroll = PaginationScroll(Order, secret=settings.scroll_secret)
@app.get("/orders/export")
async def export(params: Annotated[ScrollParams, Depends(scroll)]) -> ScrollPage[Order]:
if params.scroll_id is None:
return await scroll.start(await find_orders(), params)
return await scroll.next(params)
The batches are what was found when the scroll started, whatever is written meanwhile. The
snapshot lasts keep_alive seconds after each batch; once it has expired, its scroll_id is a
404 and the client starts again. A scroll_id is signed (HMAC-SHA256) and holds the
snapshot's id and where the next batch starts, so a retried request gets the same batch.
The snapshot is one entry under Cache.state_key, which the cache's clear() leaves alone,
in the shared tier of a TieredBackend, and sealed with the cache's signing_key or
encryption_key (the latter when it holds personal data). Whoever has a scroll_id reads the
rest of the snapshot without the endpoint's query running again: when the results depend on
who asks, pass owner (the user's id) to start and next, and a scroll_id is only good
for them. Each batch reads the whole snapshot from the backend and writes it back to renew
it: keep max_items modest.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
item_type
|
type[T]
|
the type of the items, which writes them into the snapshot as JSON and reads them back (a pydantic model, a dataclass...). |
required |
secret
|
Secrets
|
at least 32 bytes ( |
required |
default_batch_size
|
int
|
items per batch when the request doesn't say. |
100
|
max_batch_size
|
int
|
the most a request may ask for (a 422 beyond). |
1000
|
keep_alive
|
int
|
seconds the snapshot lasts after its last batch. |
60
|
max_items
|
int
|
the most a snapshot holds: more is a ValueError at |
10000
|
cache
|
Cache | None
|
the Cache whose backend keeps the snapshots; FastAPICache's by default, resolved at each call. |
None
|
namespace
|
str
|
part of the keys, after the cache's prefix (see |
'scroll'
|
params
¶
params(*, scroll_id: str | None = None, batch_size: int | None = None) -> ScrollParams
The parameters of a batch, with the scroll_id decoded, as the dependency gives them. Raises an HTTPException (400) if the scroll_id isn't valid.
start
async
¶
start(items: Iterable[T], params: ScrollParams, *, owner: str | None = None) -> ScrollPage[T]
Keep items as the snapshot of a new scroll, for keep_alive seconds, and return its
first batch. When they fit in it, nothing is kept and the scroll is over.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
items
|
Iterable[T]
|
what the query found, in order: a list, or rows read as they come. More
than |
required |
owner
|
str | None
|
who may read the rest (a user's id): |
None
|
next
async
¶
next(params: ScrollParams, *, owner: str | None = None) -> ScrollPage[T]
The batch that params.scroll_id points at, with the snapshot's lifetime renewed.
Raises an HTTPException: 404 when the snapshot has expired (or isn't one of this
owner), 503 when the backend fails.
clear
async
¶
clear(scroll_id: str, *, owner: str | None = None) -> bool
Delete the snapshot of scroll_id before it expires, when the client is done with
it; whether it was there. Raises an HTTPException: 400 if the scroll_id isn't valid,
503 when the backend fails.
fastapi_alcazaba.pagination.streaming.PaginationStreaming
¶
PaginationStreaming(*, batch_size: int = 100, max_items: int | None = None)
Every item, as they come, without pages: not a dependency, but a response.
stream = PaginationStreaming(batch_size=100)
@app.get("/events")
async def events() -> StreamingResponse:
async def fetch(after: int | None, limit: int) -> list[Event]:
... # WHERE id > :after ORDER BY id LIMIT :limit
return stream.response(stream.batches(fetch, key=lambda e: e.id))
The response is NDJSON (application/x-ndjson): one JSON value per line, sent in chunks of
batch_size lines, so that a client starts reading while the rest is being queried. A
client that disconnects stops it. Nothing is cached: @cache doesn't keep streams.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
batch_size
|
int
|
lines in each chunk of the response, and rows in each query of
|
100
|
max_items
|
int | None
|
the most items a response sends: it ends there, whatever is left. None for no end. |
None
|
response
¶
response(items: Iterable[Any] | AsyncIterable[Any]) -> StreamingResponse
A response that sends items as NDJSON, models serialised with pydantic.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
items
|
Iterable[Any] | AsyncIterable[Any]
|
a list, a generator, or an async iterable (an async generator,
|
required |
batches
async
¶
batches(fetch: Callable[[K | None, int], Iterable[T] | Awaitable[Iterable[T]]], *, key: Callable[[T], K]) -> AsyncIterator[T]
Every row of a big table, read batch_size rows at a time by key, without loading it
whole: fetch(None, batch_size) first, then fetch(key(last row), batch_size), until a
batch comes back shorter than batch_size. Each query finds its batch through an index,
as fast as the first, and rows written meanwhile don't shift the batches.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fetch
|
Callable[[K | None, int], Iterable[T] | Awaitable[Iterable[T]]]
|
|
required |
key
|
Callable[[T], K]
|
a row's sort key, unique ( Raises ValueError if |
required |
fastapi_alcazaba.pagination.cursor.CursorPage
¶
Bases: LinkedPage, Generic[T]
A page of items by cursor: prev_cursor asks for the page before it (None on the
first), next_cursor for the next (None on the last).
fastapi_alcazaba.pagination.cursor.CursorParams
dataclass
¶
CursorParams(limit: int = 50, cursor: str | None = None, after: Position | None = None, before: Position | None = None, query_digest: str = _NO_QUERY, links: PageLinks | None = None)
Bases: PaginationParams
What PaginationCursor reads from the query: limit and the cursor (none for the first
page), with the position it points at already decoded: after for a next page, before
for a previous one. query_digest is the digest of the request's other query parameters,
which the page's cursors are bound to; it's part of the cache key, so that each query has its
own pages and its own cursors.
fetch
property
¶
fetch: int
Rows to query from the database: limit and one more, which tells whether there's
another page in the direction of the query.
fastapi_alcazaba.pagination.keyset.KeysetPage
¶
Bases: LinkedPage, Generic[T]
A page of items by keyset: after is the sort key of its last item, where the next page
starts (None when it's empty), and has_more is False on the last one. A client that syncs
keeps after from the last page, and asks with it later for what came since.
fastapi_alcazaba.pagination.keyset.KeysetParams
dataclass
¶
KeysetParams(limit: int = 50, after: Position | None = None, links: PageLinks | None = None)
Bases: PaginationParams
What PaginationKeyset reads from the query: limit, and after, the sort key of the
last item seen, converted to its types (none for the first page): a value, or a tuple of
them when the endpoint sorts by several.
fetch
property
¶
fetch: int
Rows to query from the database: limit and one more, which tells whether there's a
next page.
fastapi_alcazaba.pagination.offset.OffsetPage
¶
Bases: LinkedPage, Generic[T]
A page of items by offset; has_more is False on the last one, and total is None when
the list wasn't counted.
fastapi_alcazaba.pagination.offset.OffsetParams
dataclass
¶
OffsetParams(offset: int = 0, limit: int = 50, links: PageLinks | None = None)
Bases: PaginationParams
What PaginationOffset reads from the query: offset and limit.
fetch
property
¶
fetch: int
Rows to query from the database without counting the list: limit and one more,
which tells whether there's a next page.
fastapi_alcazaba.pagination.page_number.PageNumberPage
¶
Bases: LinkedPage, Generic[T]
A page of items by page number: PageNumberPage[Product] as the endpoint's return type.
fastapi_alcazaba.pagination.page_number.PageNumberParams
dataclass
¶
PageNumberParams(page: int = 1, page_size: int = 50, links: PageLinks | None = None)
Bases: PaginationParams
What PaginationPageNumber reads from the query: page (from 1) and page_size.
fastapi_alcazaba.pagination.ranges.RangePage
¶
Bases: LinkedPage, Generic[T]
A range of items, with the total, which the response repeats in Content-Range.
fastapi_alcazaba.pagination.ranges.RangeParams
dataclass
¶
RangeParams(start: int = 0, end: int = 49, links: PageLinks | None = None)
Bases: PaginationParams
What PaginationRange reads from the query: start and end, the indexes of the first
and last items, from 0 and both included.
fastapi_alcazaba.pagination.scroll.ScrollPage
¶
fastapi_alcazaba.pagination.scroll.ScrollParams
dataclass
¶
ScrollParams(batch_size: int = 100, scroll_id: str | None = None, snapshot: str | None = None, offset: int = 0, links: PageLinks | None = None)
Bases: PaginationParams
What PaginationScroll reads from the query: batch_size and the scroll_id (none to
start a scroll), with what it holds already decoded: the snapshot it reads and the
offset where its next batch starts.
fastapi_alcazaba.pagination.sqlalchemy.page_query
¶
page_query(statement: SelectT, params: OffsetParams | PageNumberParams | RangeParams | CursorParams | KeysetParams, *, by: Sequence[Column] = (), descending: bool = False) -> SelectT
statement limited to the page of params.
By position (OffsetParams, PageNumberParams, RangeParams): OFFSET and LIMIT, in the
order statement already has, or the one by gives. OffsetParams fetch one row more,
which PaginationOffset.paginate(..., queried=True) takes to tell whether there's a next
page; with total= it leaves the extra row out.
By key (CursorParams, KeysetParams): by is the sort key, unique as a whole ((Event.
created_at, Event.id)), and the statement must have no ORDER BY of its own. The rows past
params.after, in the order of by (descending for newest first), or for a previous page
those before params.before in the reverse order, nearest first, as paginate(...,
queried=True) expects them; one more than the limit in either case.
fastapi_alcazaba.pagination.sqlalchemy.count_query
¶
count_query(statement: Select[Any]) -> Select[int]
The count of the rows of statement, without its order, limit or offset: the total of
PaginationPageNumber, PaginationOffset and PaginationRange.
fastapi_alcazaba.pagination.sqlalchemy.sort_key
¶
sort_key(*columns: Column) -> Callable[[Any], Position]
The key of paginate for these columns: a row's values of them, read by their names
(sort_key(Event.created_at, Event.id) gives (event.created_at, event.id)); one column's
value alone.
fastapi_alcazaba.pagination.PaginationParams
¶
The base of what each strategy reads from the query. The cache looks for it among the endpoint's arguments, to give a page served from it the Link header of the request it answers.
fastapi_alcazaba.pagination.LinkedPage
¶
Bases: BaseModel
The base of the pages: relations() says which pages the Link header points at, and
headers() what else the response carries (Content-Range...).
fastapi_alcazaba.pagination.PageLinks
dataclass
¶
PageLinks(response: Response, path: str, query: tuple[tuple[str, str], ...])
Where the Link header of a page goes: the response to the request that asked for it, and
the request's path and query parameters. The links keep the other parameters (filters,
sorting) and are relative to the host (</products?page=3&page_size=20>), so they don't
depend on the Host header nor on whether a proxy in front speaks HTTPS.
add
¶
add(page: object) -> None
Add the links of page (its relations()) to the response's Link header, and its
other headers().