|
| 1 | +--- |
| 2 | +name: fastsqla-pagination |
| 3 | +description: > |
| 4 | + Paginate SQLAlchemy select queries in FastAPI endpoints using FastSQLA. |
| 5 | + Covers the built-in Paginate dependency (offset/limit query params), |
| 6 | + Page/Item/Collection response models, and the new_pagination() factory |
| 7 | + for custom page sizes, count queries, and result processing. |
| 8 | +--- |
| 9 | + |
| 10 | +# FastSQLA Pagination |
| 11 | + |
| 12 | +FastSQLA provides a `Paginate` dependency that adds `offset` and `limit` query parameters to any FastAPI endpoint and returns paginated results wrapped in a `Page` model. |
| 13 | + |
| 14 | +## Response Models |
| 15 | + |
| 16 | +FastSQLA exports three generic response wrappers: |
| 17 | + |
| 18 | +### `Page[T]` — paginated list with metadata |
| 19 | + |
| 20 | +```json |
| 21 | +{ |
| 22 | + "data": [{ ... }, { ... }], |
| 23 | + "meta": { |
| 24 | + "offset": 0, |
| 25 | + "total_items": 42, |
| 26 | + "total_pages": 5, |
| 27 | + "page_number": 1 |
| 28 | + } |
| 29 | +} |
| 30 | +``` |
| 31 | + |
| 32 | +### `Collection[T]` — plain list, no pagination metadata |
| 33 | + |
| 34 | +```json |
| 35 | +{ |
| 36 | + "data": [{ ... }, { ... }] |
| 37 | +} |
| 38 | +``` |
| 39 | + |
| 40 | +### `Item[T]` — single item wrapper |
| 41 | + |
| 42 | +```json |
| 43 | +{ |
| 44 | + "data": { ... } |
| 45 | +} |
| 46 | +``` |
| 47 | + |
| 48 | +## Basic Usage with `Paginate` |
| 49 | + |
| 50 | +`Paginate` is a pre-configured FastAPI dependency. It injects a callable that accepts a SQLAlchemy `Select` and returns a `Page`. |
| 51 | + |
| 52 | +Default query parameters added to the endpoint: |
| 53 | +- `offset`: int, default `0`, minimum `0` |
| 54 | +- `limit`: int, default `10`, minimum `1`, maximum `100` |
| 55 | + |
| 56 | +```python |
| 57 | +from fastapi import FastAPI |
| 58 | +from fastsqla import Base, Page, Paginate |
| 59 | +from pydantic import BaseModel |
| 60 | +from sqlalchemy import select |
| 61 | +from sqlalchemy.orm import Mapped, mapped_column |
| 62 | + |
| 63 | +app = FastAPI() |
| 64 | + |
| 65 | +class Hero(Base): |
| 66 | + __tablename__ = "hero" |
| 67 | + id: Mapped[int] = mapped_column(primary_key=True) |
| 68 | + name: Mapped[str] = mapped_column(unique=True) |
| 69 | + age: Mapped[int] |
| 70 | + |
| 71 | +class HeroModel(BaseModel): |
| 72 | + id: int |
| 73 | + name: str |
| 74 | + age: int |
| 75 | + |
| 76 | +@app.get("/heroes") |
| 77 | +async def list_heroes(paginate: Paginate[HeroModel]) -> Page[HeroModel]: |
| 78 | + return await paginate(select(Hero)) |
| 79 | +``` |
| 80 | + |
| 81 | +A request to `GET /heroes?offset=20&limit=10` returns the third page of results. |
| 82 | + |
| 83 | +## Adding Filters |
| 84 | + |
| 85 | +Combine `Paginate` with additional query parameters: |
| 86 | + |
| 87 | +```python |
| 88 | +@app.get("/heroes") |
| 89 | +async def list_heroes( |
| 90 | + paginate: Paginate[HeroModel], |
| 91 | + age: int | None = None, |
| 92 | + name: str | None = None, |
| 93 | +): |
| 94 | + stmt = select(Hero) |
| 95 | + if age is not None: |
| 96 | + stmt = stmt.where(Hero.age == age) |
| 97 | + if name is not None: |
| 98 | + stmt = stmt.where(Hero.name.ilike(f"%{name}%")) |
| 99 | + return await paginate(stmt) |
| 100 | +``` |
| 101 | + |
| 102 | +## The `new_pagination()` Factory |
| 103 | + |
| 104 | +For custom pagination behavior, use `new_pagination()` to create a new dependency. It accepts four parameters: |
| 105 | + |
| 106 | +| Parameter | Type | Default | Description | |
| 107 | +|-----------|------|---------|-------------| |
| 108 | +| `min_page_size` | `int` | `10` | Default and minimum `limit` value | |
| 109 | +| `max_page_size` | `int` | `100` | Maximum allowed `limit` value | |
| 110 | +| `query_count_dependency` | `Callable[..., Awaitable[int]] \| None` | `None` | FastAPI dependency returning total item count. When `None`, uses `SELECT COUNT(*) FROM (subquery)`. | |
| 111 | +| `result_processor` | `Callable[[Result], Iterable]` | `lambda r: iter(r.unique().scalars())` | Transforms the SQLAlchemy `Result` into an iterable of items | |
| 112 | + |
| 113 | +The return value is a FastAPI dependency. Use it with `Annotated` and `Depends`: |
| 114 | + |
| 115 | +```python |
| 116 | +from typing import Annotated |
| 117 | +from fastapi import Depends |
| 118 | +from fastsqla import PaginateType, new_pagination |
| 119 | +``` |
| 120 | + |
| 121 | +### `PaginateType[T]` |
| 122 | + |
| 123 | +Type alias for the paginate callable: |
| 124 | + |
| 125 | +```python |
| 126 | +type PaginateType[T] = Callable[[Select], Awaitable[Page[T]]] |
| 127 | +``` |
| 128 | + |
| 129 | +Use this when annotating custom pagination dependencies. |
| 130 | + |
| 131 | +## Custom Page Sizes |
| 132 | + |
| 133 | +```python |
| 134 | +from typing import Annotated |
| 135 | +from fastapi import Depends |
| 136 | +from fastsqla import Page, PaginateType, new_pagination |
| 137 | + |
| 138 | +SmallPagePaginate = Annotated[ |
| 139 | + PaginateType[HeroModel], |
| 140 | + Depends(new_pagination(min_page_size=5, max_page_size=25)), |
| 141 | +] |
| 142 | + |
| 143 | +@app.get("/heroes") |
| 144 | +async def list_heroes(paginate: SmallPagePaginate) -> Page[HeroModel]: |
| 145 | + return await paginate(select(Hero)) |
| 146 | +``` |
| 147 | + |
| 148 | +This endpoint has `limit` defaulting to `5` with a maximum of `25`. |
| 149 | + |
| 150 | +## Custom Count Query |
| 151 | + |
| 152 | +The default count query runs `SELECT COUNT(*) FROM (your_select_as_subquery)`. For joins or complex queries where this is inefficient, provide a `query_count_dependency` — a FastAPI dependency that receives the session and returns an `int`: |
| 153 | + |
| 154 | +```python |
| 155 | +from typing import cast |
| 156 | +from sqlalchemy import func, select |
| 157 | +from fastsqla import Session |
| 158 | + |
| 159 | +async def query_count(session: Session) -> int: |
| 160 | + result = await session.execute(select(func.count()).select_from(Sticky)) |
| 161 | + return cast(int, result.scalar()) |
| 162 | +``` |
| 163 | + |
| 164 | +Then pass it to `new_pagination()`: |
| 165 | + |
| 166 | +```python |
| 167 | +CustomPaginate = Annotated[ |
| 168 | + PaginateType[StickyModel], |
| 169 | + Depends(new_pagination(query_count_dependency=query_count)), |
| 170 | +] |
| 171 | +``` |
| 172 | + |
| 173 | +### Count Query with Filters |
| 174 | + |
| 175 | +Since `query_count_dependency` is a FastAPI dependency, it can accept query parameters and other dependencies. This is useful when the count must reflect the same filters applied to the main query: |
| 176 | + |
| 177 | +```python |
| 178 | +from sqlalchemy import func, select |
| 179 | +from fastsqla import Session |
| 180 | + |
| 181 | +async def filtered_hero_count( |
| 182 | + session: Session, |
| 183 | + age: int | None = None, |
| 184 | + name: str | None = None, |
| 185 | +) -> int: |
| 186 | + stmt = select(func.count()).select_from(Hero) |
| 187 | + if age is not None: |
| 188 | + stmt = stmt.where(Hero.age == age) |
| 189 | + if name is not None: |
| 190 | + stmt = stmt.where(Hero.name.ilike(f"%{name}%")) |
| 191 | + result = await session.execute(stmt) |
| 192 | + return cast(int, result.scalar()) |
| 193 | + |
| 194 | +FilteredPaginate = Annotated[ |
| 195 | + PaginateType[HeroModel], |
| 196 | + Depends(new_pagination(query_count_dependency=filtered_hero_count)), |
| 197 | +] |
| 198 | + |
| 199 | +@app.get("/heroes") |
| 200 | +async def list_heroes( |
| 201 | + paginate: FilteredPaginate, |
| 202 | + age: int | None = None, |
| 203 | + name: str | None = None, |
| 204 | +) -> Page[HeroModel]: |
| 205 | + stmt = select(Hero) |
| 206 | + if age is not None: |
| 207 | + stmt = stmt.where(Hero.age == age) |
| 208 | + if name is not None: |
| 209 | + stmt = stmt.where(Hero.name.ilike(f"%{name}%")) |
| 210 | + return await paginate(stmt) |
| 211 | +``` |
| 212 | + |
| 213 | +FastAPI resolves the shared `age` and `name` query parameters in both the endpoint and the count dependency, so the count always matches the filtered results. |
| 214 | + |
| 215 | +## Custom Result Processor |
| 216 | + |
| 217 | +The default `result_processor` is: |
| 218 | + |
| 219 | +```python |
| 220 | +lambda result: iter(result.unique().scalars()) |
| 221 | +``` |
| 222 | + |
| 223 | +This works for single-entity selects like `select(Hero)`. For multi-column selects (e.g., joins returning individual columns), use `.mappings()`: |
| 224 | + |
| 225 | +```python |
| 226 | +lambda result: iter(result.mappings()) |
| 227 | +``` |
| 228 | + |
| 229 | +## Full Custom Pagination Example |
| 230 | + |
| 231 | +Combining a custom count query and a custom result processor for a join: |
| 232 | + |
| 233 | +```python |
| 234 | +from typing import Annotated, cast |
| 235 | +from fastapi import Depends, FastAPI |
| 236 | +from fastsqla import Base, Page, PaginateType, Session, new_pagination |
| 237 | +from pydantic import BaseModel |
| 238 | +from sqlalchemy import ForeignKey, String, func, select |
| 239 | +from sqlalchemy.orm import Mapped, mapped_column |
| 240 | + |
| 241 | +app = FastAPI() |
| 242 | + |
| 243 | +class User(Base): |
| 244 | + __tablename__ = "user" |
| 245 | + id: Mapped[int] = mapped_column(primary_key=True) |
| 246 | + email: Mapped[str] = mapped_column(String, unique=True) |
| 247 | + name: Mapped[str] |
| 248 | + |
| 249 | +class Sticky(Base): |
| 250 | + __tablename__ = "sticky" |
| 251 | + id: Mapped[int] = mapped_column(primary_key=True) |
| 252 | + user_id: Mapped[int] = mapped_column(ForeignKey(User.id)) |
| 253 | + body: Mapped[str] |
| 254 | + |
| 255 | +class StickyModel(BaseModel): |
| 256 | + id: int |
| 257 | + body: str |
| 258 | + user_id: int |
| 259 | + user_email: str |
| 260 | + user_name: str |
| 261 | + |
| 262 | +async def query_count(session: Session) -> int: |
| 263 | + result = await session.execute(select(func.count()).select_from(Sticky)) |
| 264 | + return cast(int, result.scalar()) |
| 265 | + |
| 266 | +CustomPaginate = Annotated[ |
| 267 | + PaginateType[StickyModel], |
| 268 | + Depends( |
| 269 | + new_pagination( |
| 270 | + query_count_dependency=query_count, |
| 271 | + result_processor=lambda result: iter(result.mappings()), |
| 272 | + ) |
| 273 | + ), |
| 274 | +] |
| 275 | + |
| 276 | +@app.get("/stickies") |
| 277 | +async def list_stickies(paginate: CustomPaginate) -> Page[StickyModel]: |
| 278 | + stmt = select( |
| 279 | + Sticky.id, |
| 280 | + Sticky.body, |
| 281 | + User.id.label("user_id"), |
| 282 | + User.email.label("user_email"), |
| 283 | + User.name.label("user_name"), |
| 284 | + ).join(User) |
| 285 | + return await paginate(stmt) |
| 286 | +``` |
| 287 | + |
| 288 | +## SQLModel Usage |
| 289 | + |
| 290 | +When using SQLModel, models serve as both ORM and response models — no separate Pydantic model is needed: |
| 291 | + |
| 292 | +```python |
| 293 | +from fastsqla import Page, Paginate |
| 294 | +from sqlmodel import Field, SQLModel, select |
| 295 | + |
| 296 | +class Hero(SQLModel, table=True): |
| 297 | + id: int | None = Field(default=None, primary_key=True) |
| 298 | + name: str = Field(unique=True) |
| 299 | + age: int |
| 300 | + |
| 301 | +@app.get("/heroes") |
| 302 | +async def list_heroes(paginate: Paginate[Hero]) -> Page[Hero]: |
| 303 | + return await paginate(select(Hero)) |
| 304 | +``` |
| 305 | + |
| 306 | +## Quick Reference |
| 307 | + |
| 308 | +| What you need | What to use | |
| 309 | +|---|---| |
| 310 | +| Standard pagination (offset/limit) | `Paginate[T]` | |
| 311 | +| Custom page sizes | `Annotated[PaginateType[T], Depends(new_pagination(min_page_size=..., max_page_size=...))]` | |
| 312 | +| Custom count for joins | `new_pagination(query_count_dependency=my_count_dep)` | |
| 313 | +| Multi-column select results | `new_pagination(result_processor=lambda r: iter(r.mappings()))` | |
| 314 | +| Type annotation for paginate callable | `PaginateType[T]` | |
| 315 | +| Paginated response | `Page[T]` (data + meta) | |
| 316 | +| Unpaginated list response | `Collection[T]` (data only) | |
| 317 | +| Single item response | `Item[T]` (data only) | |
0 commit comments