Home/Learn/FastAPI/Pagination, Filtering & Sorting — List Endpoints Done Right

Pagination, Filtering & Sorting — List Endpoints Done Right

Intermediate

Databases

Production list endpoints return an envelope (items + total + page), build filters dynamically but safely, whitelist sort fields against injection, and switch to keyset pagination when OFFSET gets slow.

Overview

Every resource grows a list endpoint, and every list endpoint eventually needs the same four things: pagination (never return unbounded lists), filtering (by branch, by CGPA range, by search term), sorting (whitelisted — clients must not order by arbitrary SQL), and an envelope carrying total count so the frontend can render page numbers. The dependency system packages all of it reusably. The scaling twist: OFFSET pagination reads and discards all skipped rows, so page 2000 of a big table is slow and shifts when rows are inserted — keyset (cursor) pagination filters on the last-seen key instead, staying fast at any depth; it is how Instagram-style infinite scroll works and a favourite systems interview question.

The Envelope: Filters + Sort Whitelist + Total

Optional query params compose into a WHERE list; the sort field maps through a whitelist dict (never raw column names from clients); one count query + one page query fill the envelope.

Envelope + dynamic WHERE + whitelisted ORDER BY

from fastapi import Depends, FastAPI, HTTPException, Query
from pydantic import BaseModel
from sqlalchemy import func, select
from sqlalchemy.orm import Session

app = FastAPI()

SORTABLE = {"cgpa": Student.cgpa, "name": Student.name, "id": Student.id}

class Page(BaseModel):
    items: list[StudentOut]
    total: int
    page: int
    size: int

@app.get("/students", response_model=Page)
def list_students(
    db: Session = Depends(get_db),
    branch: str | None = None,
    min_cgpa: float = Query(default=0, ge=0, le=10),
    q: str | None = Query(default=None, min_length=2),     # name search
    sort: str = Query(default="id"),
    order: str = Query(default="asc", pattern="^(asc|desc)$"),
    page: int = Query(default=1, ge=1),
    size: int = Query(default=20, ge=1, le=100),           # hard cap, always
):
    if sort not in SORTABLE:                # whitelist — NOT getattr(Student, sort)
        raise HTTPException(422, f"sort must be one of {sorted(SORTABLE)}")

    filters = [Student.cgpa >= min_cgpa]
    if branch:
        filters.append(Student.branch == branch)
    if q:
        filters.append(Student.name.ilike(f"%{q}%"))       # parameterised — safe

    col = SORTABLE[sort]
    stmt = (select(Student).where(*filters)
            .order_by(col.desc() if order == "desc" else col.asc())
            .offset((page - 1) * size).limit(size))

    total = db.execute(select(func.count()).select_from(Student).where(*filters)).scalar()
    items = db.execute(stmt).scalars().all()
    return Page(items=items, total=total, page=page, size=size)

# GET /students?branch=CS&min_cgpa=8&sort=cgpa&order=desc&page=2&size=25
# → {"items":[...25...], "total":312, "page":2, "size":25}

Keyset (Cursor) Pagination — Deep Pages Without OFFSET

OFFSET 100000 reads and throws away 100k rows, and page boundaries drift as data changes. Keyset filters on the last-seen key: index-fast at any depth, stable under inserts — the infinite-scroll standard.

WHERE id < last_seen beats OFFSET at every depth

# The problem, in SQL terms:
#   OFFSET 100000 LIMIT 20  → scans 100,020 rows, returns 20  (and slower
#   every page); a row inserted meanwhile shifts every later page.

# Keyset: remember where you stopped, filter past it.
import base64

@app.get("/feed/offers")
def offers_feed(
    db: Session = Depends(get_db),
    cursor: str | None = None,               # opaque to clients
    size: int = Query(default=20, ge=1, le=100),
):
    stmt = select(Offer).order_by(Offer.id.desc()).limit(size + 1)  # +1 = "has more?"
    if cursor:
        last_id = int(base64.urlsafe_b64decode(cursor).decode())
        stmt = stmt.where(Offer.id < last_id)     # index seek, not scan

    rows = db.execute(stmt).scalars().all()
    has_more, page_rows = len(rows) > size, rows[:size]

    next_cursor = None
    if has_more:
        next_cursor = base64.urlsafe_b64encode(
            str(page_rows[-1].id).encode()).decode()

    return {
        "items": [{"id": o.id, "company": o.company, "ctc_lpa": o.ctc_lpa}
                  for o in page_rows],
        "next_cursor": next_cursor,               # null = end of feed
    }
# Client: GET /feed/offers → GET /feed/offers?cursor=MTgz → ...
#
# Trade-offs vs OFFSET:
#   ✓ constant-time at any depth (index seek)   ✓ stable under inserts
#   ✗ no "jump to page 47"  ✗ sort key must be unique-ish (tie-break with id)
# Rule: admin tables with page numbers → OFFSET (capped);
#       feeds / infinite scroll / exports → keyset.

Key Points to Remember

1Always cap size (le=100) and return an envelope: items + total + page info
2Sort fields go through a whitelist dict — never client-supplied column names
3Filters compose as a WHERE list; ilike with bound params is injection-safe
4OFFSET degrades with depth and drifts; keyset (WHERE key < cursor) stays fast and stable

Interview Questions

Why is OFFSET pagination slow on page 2000, and what replaces it?

HardFlipkart

A client sends ?sort=password_hash — how does your endpoint stay safe?

MediumPostman

Design the pagination for an infinite-scroll job feed: response shape, cursor contents, tie-breaking.

HardAmazon

Ask Aria about Pagination, Filtering & Sorting — List Endpoints Done Right

Your personal AI tutor — ask anything about this concept

Revision Status

Personal Notes

Discussion

Loading discussion…

Async SQLAlchemy — asyncpg and AsyncSession

OAuth2 Password Flow — Login That Issues Tokens