Pagination

fastsqla.Paginate[T]

A dependency used in endpoints to paginate SQLAlchemy select queries.

It adds offsetand limit query parameters to the endpoint, which are used to paginate. The model returned by the endpoint is a Page model. It contains a page of data and metadata:

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

---

Example:

example.py

from fastsqla import Base, Paginate, Page
from pydantic import BaseModel


class Hero(Base):
    __tablename__ = "hero"


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(HeroBase):
    model_config = ConfigDict(from_attributes=True)
    id: int


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

1. The endpoint returns a Page model of HeroModel.
2. Just define an argument with type Paginate to get an async paginate function injected in your endpoint function.
3. 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:

from fastsqla import Paginate, Page

@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)