Cheatsheet for pagination in FastAPI.
Offset/limit (simple)
@app.get("/users")
async def list_(
limit: int = Query(20, ge=1, le=100),
offset: int = Query(0, ge=0),
db: AsyncSession = Depends(get_db),
):
rows = (await db.execute(
select(User).order_by(User.id).limit(limit).offset(offset)
)).scalars().all()
return {"items": rows, "limit": limit, "offset": offset}
Simple but slow at scale (offset walks rows).
Keyset (recommended)
@app.get("/users")
async def list_(
after_id: int = Query(0, ge=0),
limit: int = Query(20, ge=1, le=100),
db: AsyncSession = Depends(get_db),
):
stmt = select(User).where(User.id > after_id).order_by(User.id).limit(limit + 1)
rows = (await db.execute(stmt)).scalars().all()
has_more = len(rows) > limit
rows = rows[:limit]
next_cursor = rows[-1].id if has_more else None
return {"items": rows, "next_cursor": next_cursor, "has_more": has_more}
Stable under inserts. O(log n) per page.
Opaque cursor
Encode keyset state opaquely:
import base64, json
def encode_cursor(d: dict) -> str:
return base64.urlsafe_b64encode(json.dumps(d).encode()).decode()
def decode_cursor(s: str) -> dict:
return json.loads(base64.urlsafe_b64decode(s).decode())
@app.get("/posts")
async def posts(
cursor: str | None = None,
limit: int = Query(20, ge=1, le=100),
db: AsyncSession = Depends(get_db),
):
after = decode_cursor(cursor) if cursor else None
stmt = select(Post).order_by(Post.created_at.desc(), Post.id.desc())
if after:
stmt = stmt.where(
tuple_(Post.created_at, Post.id) < (after["created_at"], after["id"])
)
rows = (await db.execute(stmt.limit(limit + 1))).scalars().all()
has_more = len(rows) > limit
rows = rows[:limit]
next_c = encode_cursor({"created_at": rows[-1].created_at.isoformat(), "id": rows[-1].id}) if has_more else None
return {"items": rows, "next_cursor": next_c, "has_more": has_more}
Use opaque cursors when you may change ordering / fields without breaking clients.
Generic Page model
from typing import TypeVar, Generic
T = TypeVar("T")
class Page(BaseModel, Generic[T]):
items: list[T]
next_cursor: str | None = None
has_more: bool
@app.get("/posts", response_model=Page[PostOut])
async def posts(...): ...
OpenAPI generates Page_PostOut_ schema. SDK clients get typed pagination.
Total counts (optional + expensive)
@app.get("/users")
async def list_(...):
items = ...
total = await db.scalar(select(func.count(User.id)))
return {"items": items, "total": total, ...}
Don’t COUNT(*) on large tables every page — use approximate counts (PG: pg_stat_user_tables.n_live_tup).
SQLAlchemy with_only_columns for COUNT
total = await db.scalar(stmt.with_only_columns(func.count()).order_by(None))
Reuses base stmt without ORDER BY (faster).
Async iter for full export
async def all_users(db):
async for u in (await db.stream(select(User).order_by(User.id))).scalars():
yield u
For exports / NDJSON streaming.
Filters + pagination
class PostFilter(BaseModel):
q: str | None = None
author: int | None = None
@app.get("/posts")
async def posts(
f: PostFilter = Depends(),
after_id: int = 0,
limit: int = Query(20, le=100),
db: AsyncSession = Depends(get_db),
):
stmt = select(Post)
if f.q: stmt = stmt.where(Post.title.ilike(f"%{f.q}%"))
if f.author: stmt = stmt.where(Post.author_id == f.author)
stmt = stmt.where(Post.id > after_id).order_by(Post.id).limit(limit + 1)
rows = (await db.execute(stmt)).scalars().all()
has_more = len(rows) > limit
rows = rows[:limit]
return Page(items=rows, next_cursor=str(rows[-1].id) if has_more else None, has_more=has_more)
Headers (alternative to body cursor)
@app.get("/users")
async def list_(response: Response, ...):
rows, next_c, has_more = ...
if next_c:
response.headers["X-Next-Cursor"] = next_c
response.headers["Link"] = f'<{request.url.include_query_params(after_id=next_c)}>; rel="next"'
return rows
GitHub-style. Some clients prefer headers; some prefer body.
Read this next
If you want my generic pagination library, it’s at rajpoot.dev .
Building something AI-, backend-, or data-heavy and want a second pair of eyes? I do consulting and freelance work — see my projects and ways to reach me at rajpoot.dev .