Skip to content

Commit 8e3cf6b

Browse files
committed
feat: add fastsqla-pagination agent skill
1 parent 6e98cf4 commit 8e3cf6b

1 file changed

Lines changed: 275 additions & 0 deletions

File tree

Lines changed: 275 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,275 @@
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+
## Custom Result Processor
174+
175+
The default `result_processor` is:
176+
177+
```python
178+
lambda result: iter(result.unique().scalars())
179+
```
180+
181+
This works for single-entity selects like `select(Hero)`. For multi-column selects (e.g., joins returning individual columns), use `.mappings()`:
182+
183+
```python
184+
lambda result: iter(result.mappings())
185+
```
186+
187+
## Full Custom Pagination Example
188+
189+
Combining a custom count query and a custom result processor for a join:
190+
191+
```python
192+
from typing import Annotated, cast
193+
from fastapi import Depends, FastAPI
194+
from fastsqla import Base, Page, PaginateType, Session, new_pagination
195+
from pydantic import BaseModel
196+
from sqlalchemy import ForeignKey, String, func, select
197+
from sqlalchemy.orm import Mapped, mapped_column
198+
199+
app = FastAPI()
200+
201+
class User(Base):
202+
__tablename__ = "user"
203+
id: Mapped[int] = mapped_column(primary_key=True)
204+
email: Mapped[str] = mapped_column(String, unique=True)
205+
name: Mapped[str]
206+
207+
class Sticky(Base):
208+
__tablename__ = "sticky"
209+
id: Mapped[int] = mapped_column(primary_key=True)
210+
user_id: Mapped[int] = mapped_column(ForeignKey(User.id))
211+
body: Mapped[str]
212+
213+
class StickyModel(BaseModel):
214+
id: int
215+
body: str
216+
user_id: int
217+
user_email: str
218+
user_name: str
219+
220+
async def query_count(session: Session) -> int:
221+
result = await session.execute(select(func.count()).select_from(Sticky))
222+
return cast(int, result.scalar())
223+
224+
CustomPaginate = Annotated[
225+
PaginateType[StickyModel],
226+
Depends(
227+
new_pagination(
228+
query_count_dependency=query_count,
229+
result_processor=lambda result: iter(result.mappings()),
230+
)
231+
),
232+
]
233+
234+
@app.get("/stickies")
235+
async def list_stickies(paginate: CustomPaginate) -> Page[StickyModel]:
236+
stmt = select(
237+
Sticky.id,
238+
Sticky.body,
239+
User.id.label("user_id"),
240+
User.email.label("user_email"),
241+
User.name.label("user_name"),
242+
).join(User)
243+
return await paginate(stmt)
244+
```
245+
246+
## SQLModel Usage
247+
248+
When using SQLModel, models serve as both ORM and response models — no separate Pydantic model is needed:
249+
250+
```python
251+
from fastsqla import Page, Paginate
252+
from sqlmodel import Field, SQLModel, select
253+
254+
class Hero(SQLModel, table=True):
255+
id: int | None = Field(default=None, primary_key=True)
256+
name: str = Field(unique=True)
257+
age: int
258+
259+
@app.get("/heroes")
260+
async def list_heroes(paginate: Paginate[Hero]) -> Page[Hero]:
261+
return await paginate(select(Hero))
262+
```
263+
264+
## Quick Reference
265+
266+
| What you need | What to use |
267+
|---|---|
268+
| Standard pagination (offset/limit) | `Paginate[T]` |
269+
| Custom page sizes | `Annotated[PaginateType[T], Depends(new_pagination(min_page_size=..., max_page_size=...))]` |
270+
| Custom count for joins | `new_pagination(query_count_dependency=my_count_dep)` |
271+
| Multi-column select results | `new_pagination(result_processor=lambda r: iter(r.mappings()))` |
272+
| Type annotation for paginate callable | `PaginateType[T]` |
273+
| Paginated response | `Page[T]` (data + meta) |
274+
| Unpaginated list response | `Collection[T]` (data only) |
275+
| Single item response | `Item[T]` (data only) |

0 commit comments

Comments
 (0)