Pagination¶
fastapi.Page[T]¶
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
Pagemodel ofHeroModel. - Just define an argument with type
Paginateto get an asyncpaginatefunction injected in your endpoint function. - Await the
paginatefunction with theSQLAlchemyselect 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)
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))