Pagination¶

`fastapi.Page[T]`¶

Bases: Collection[T]

Generic container that contains collection data and page metadata.

The Page model is used to return paginated data in paginated endpoints:

{
    "data": list[T],
    "meta": {
        "offset": int,
        "total_items": int,
        "total_pages": int,
        "page_number": int,
    }
}

`fastsqla.Paginate`¶

A dependency used in endpoints to paginate SQLAlchemy select queries.

It adds offset and limit query parameters to the endpoint, which are used to paginate. The model returned by the endpoint is a Page model.

`SQLAlchemy` example¶

example.py

from fastapi import FastAPI
from fastsqla import Base, Paginate, Page, lifespan
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.orm import Mapped, mapped_column

app = FastAPI(lifespan=lifespan)

class Hero(Base):
    __tablename__ = "hero"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)
    secret_identity: Mapped[str]
    age: Mapped[int]


class HeroModel(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str
    secret_identity: str
    age: int


@app.get("/heros", response_model=Page[HeroModel]) # (1)!
async def list_heros(paginate: Paginate): # (2)!
    return await paginate(select(Hero)) # (3)!

The endpoint returns a Page model of HeroModel.
Just define an argument with type Paginate to get an async paginate function injected in your endpoint function.
Await the paginate function with the SQLAlchemy select statement to get the paginated result.

To add filtering, just add whatever query parameters you need to the endpoint:

@app.get("/heros", response_model=Page[HeroModel])
async def list_heros(paginate: Paginate, age:int | None = None):
    stmt = select(Hero)
    if age:
        stmt = stmt.where(Hero.age == age)
    return await paginate(stmt)

Forward-only cursor pagination¶

Use cursor pagination to load the next batch of results, such as a "Load more" list. Each response includes a cursor pointing after the last returned item.

Import Page and Paginate from fastsqla.cursor. The top-level imports use offset/limit pagination. Using Hero and HeroModel from the example above:

from fastsqla.cursor import Page, Paginate

@app.get("/heroes")
async def list_heroes(paginate: Paginate[Hero]) -> Page[HeroModel]: # (1)!
    return await paginate(select(Hero).order_by(Hero.id)) # (2)!

Paginate adds optional next_cursor and limit query parameters. The default page size is 10, with a maximum of 1000.
Order by a unique, non-null column so each item has a definite position.

Request /heroes for the first page. The response contains data and meta.next_cursor. Pass that cursor as the next request's next_cursor query parameter. When next_cursor is null, there are no more results.

Filters in a JSON body¶

Query pagination also works on POST endpoints. Keep filters in the request model and use new_pagination() to configure the injectable Paginate type:

from typing import Literal
from fastsqla.cursor import Page, new_pagination
from pydantic import BaseModel, ConfigDict, Field

Paginate = new_pagination(default_page_size=5, max_page_size=50)

class HeroSearch(BaseModel):
    model_config = ConfigDict(extra="forbid")
    min_age: int | None = Field(None, ge=0)
    order_by: Literal["age", "name"] = "age"

@app.post("/heroes/search")
async def search_heroes(body: HeroSearch, paginate: Paginate[Hero]) -> Page[HeroModel]:
    column = {"age": Hero.age, "name": Hero.name}[body.order_by]
    stmt = select(Hero).order_by(column, Hero.id) # (1)!
    if body.min_age is not None:
        stmt = stmt.where(Hero.age >= body.min_age)
    return await paginate(stmt)

Hero.id breaks ties when several heroes have the same age or name.

Send this body to POST /heroes/search?limit=10:

{"min_age": 18, "order_by": "name"}

To continue, add the returned cursor to the query string. Keep the filters and ordering unchanged; omit the cursor to start a different search. Invalid cursors return HTTP 422 Unprocessable Content.

Custom parameter dependency¶

Supply a FastAPI dependency to choose where pagination parameters come from. For example, use after and size as query parameter names:

from fastapi import Query
from fastsqla.cursor import PaginationParameters, new_pagination

async def get_parameters(
    after: str | None = Query(None),
    size: int | None = Query(None),
) -> PaginationParameters:
    return {"cursor_name": "after", "cursor": after, "limit": size}

Paginate = new_pagination(
    default_page_size=5,
    max_page_size=50,
    parameters_dependency=get_parameters,
)

Use this Paginate[Hero] in the endpoint signature. The dependency must be async and return a cursor name, cursor value, and limit. A None limit uses the configured default; limits outside 1..max_page_size return HTTP 422 Unprocessable Content. The cursor name becomes the response metadata key: pass meta.after as ?after=... to continue. The final page contains "meta": {"after": null}.

Choosing a query¶

Order by non-null columns with a unique tie-breaker for the whole result set. Ascending, descending, and mixed directions are supported. Prefer immutable keys: changing a sort value during traversal can skip or repeat an item.

Supported queries

Use entity or column selections without outer joins, grouping, DISTINCT, unions, or existing limits/offsets. Ordering expressions, nullable keys, and SQLite decimal ordering are unsupported. SQLite can round decimal values when reading them, which loses the exact cursor position.

For column selections, set row_mapper=lambda row: row._mapping on new_pagination(). The default mapper returns the first selected entity or value. Each SQL row must produce one response item; filtering or deduplicating rows in the mapper breaks pagination.

Cursors contain readable ordering values. Apply authorization on every request and keep sensitive fields out of the ordering keys.

`SQLModel` example¶

from fastapi import FastAPI
from fastsqla import Page, Paginate, Session
from sqlmodel import Field, SQLModel
from sqlalchemy import select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_identity: str
    age: int


@app.get("/heroes", response_model=Page[Hero])
async def get_heroes(paginate: Paginate):
    return await paginate(select(Hero))

Pagination¶

fastapi.Page[T]¶

fastsqla.Paginate¶

SQLAlchemy example¶