Chapter 20: Pagination — Loading More Data with Query Parameters

Let’s Start With a Real Problem

Your GET /products/ endpoint has worked flawlessly since Chapter 8. Then your catalog grows — 500 products, then 50,000. One day a mobile user on a slow connection tries to open the product list, and the app just… hangs. You look at your logs: that one endpoint is now returning a 40MB JSON response and taking twelve seconds to generate. Nobody designed it to fail this way — it simply never occurred to anyone that SELECT * FROM products would eventually mean fifty thousand rows, every single time, for every single request.

Pagination is the fix: instead of returning everything at once, the client asks for a manageable slice of data at a time. In this chapter, you’ll build both of the two pagination styles you’ll actually encounter in real projects — simple offset/limit pagination, and the more scalable cursor-based pagination used by APIs handling genuinely large datasets.


What You’ll Learn

  • Why returning “everything” is a scaling problem waiting to happen
  • How to build classic skip/limit pagination with validated query parameters
  • Why offset-based pagination gets slower as a table grows, and what to do instead
  • How to build cursor-based pagination, the pattern behind most “infinite scroll” feeds
  • Theory + hands-on assignment questions to lock in what you’ve learned

Theory: Two Styles of Pagination

Offset / LimitCursor-Based
How it works“Skip the first N rows, then give me the next M”“Give me M rows after this specific position”
Client sendsskip=20&limit=10cursor=87&limit=10
Good forTraditional numbered pages (“Page 1, 2, 3…”)Infinite scroll, very large or frequently-changing tables
DownsideGets slower as skip grows — the database still has to count past all those skipped rowsCan’t easily jump to “page 5” — you can only move forward from a known position

Both are legitimate, widely-used approaches — the right choice depends on your use case, not on one being universally “better.” A searchable admin table with page numbers is a great fit for offset/limit. A social media feed that just keeps loading as you scroll is a great fit for cursor-based pagination.


Program 1: Classic Offset/Limit Pagination

Program Objective

Build a small, standalone FastAPI project with a GET /products/ endpoint supporting skip and limit query parameters, returning both the requested slice of data and useful metadata (total, skip, limit) so the client can build page numbers.

Step-by-Step Implementation

Step 1: Create the project

uv init offset-pagination-demo
cd offset-pagination-demo
uv add fastapi --extra standard
uv add pydantic-settings
uv add sqlalchemy

Step 2: Create the project structure

mkdir -p app/core app/db app/models app/schemas app/services app/routers
touch app/__init__.py app/core/__init__.py app/db/__init__.py app/models/__init__.py app/schemas/__init__.py app/services/__init__.py app/routers/__init__.py
offset-pagination-demo/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── core/
│   │   ├── __init__.py
│   │   └── config.py
│   ├── db/
│   │   ├── __init__.py
│   │   └── database.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── product.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── product.py
│   ├── services/
│   │   ├── __init__.py
│   │   └── product_service.py
│   └── routers/
│       ├── __init__.py
│       └── products.py
├── pyproject.toml
└── uv.lock

Step 3: Create each file with the code below. Step 4: Run and test.

Complete Working Source Code

app/core/config.py

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    app_name: str = "Offset Pagination Demo"
    app_version: str = "1.0.0"
    database_url: str = "sqlite:///./app.db"


settings = Settings()

app/db/database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

from app.core.config import settings

engine = create_engine(settings.database_url, connect_args={"check_same_thread": False})

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app/models/product.py

from sqlalchemy import Numeric, String
from sqlalchemy.orm import Mapped, mapped_column

from app.db.database import Base


class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(100), index=True)
    price: Mapped[float] = mapped_column(Numeric(10, 2))

app/schemas/product.py

from pydantic import BaseModel, ConfigDict


class ProductOut(BaseModel):
    id: int
    name: str
    price: float

    model_config = ConfigDict(from_attributes=True)


class PaginatedProducts(BaseModel):
    items: list[ProductOut]
    total: int
    skip: int
    limit: int

app/services/product_service.py

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.models.product import Product


def seed_products_if_empty(db: Session) -> None:
    existing_count = db.scalar(select(func.count()).select_from(Product))
    if existing_count and existing_count > 0:
        return

    for i in range(1, 51):
        db.add(Product(name=f"Product {i:02d}", price=9.99 + i))
    db.commit()


def get_paginated_products(db: Session, skip: int, limit: int) -> tuple[list[Product], int]:
    total = db.scalar(select(func.count()).select_from(Product))
    items = (
        db.query(Product)
        .order_by(Product.id)
        .offset(skip)
        .limit(limit)
        .all()
    )
    return items, total

app/routers/products.py

from typing import Annotated

from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session

from app.db.database import get_db
from app.schemas.product import PaginatedProducts
from app.services.product_service import get_paginated_products

router = APIRouter(prefix="/products", tags=["Products"])


@router.get("/", response_model=PaginatedProducts)
def list_products(
    db: Session = Depends(get_db),
    skip: Annotated[int, Query(ge=0, description="Number of records to skip")] = 0,
    limit: Annotated[int, Query(ge=1, le=100, description="Max records to return")] = 10,
):
    items, total = get_paginated_products(db, skip, limit)
    return PaginatedProducts(items=items, total=total, skip=skip, limit=limit)

app/main.py

from fastapi import FastAPI

from app.core.config import settings
from app.db.database import Base, SessionLocal, engine
from app.models import product  # noqa: F401
from app.routers import products
from app.services.product_service import seed_products_if_empty

Base.metadata.create_all(bind=engine)

with SessionLocal() as db:
    seed_products_if_empty(db)

app = FastAPI(title=settings.app_name, version=settings.app_version)

app.include_router(products.router)


@app.get("/", tags=["Root"])
def read_root():
    return {"message": f"Welcome to {settings.app_name}"}

Detailed Code Explanation

  • seed_products_if_empty exists purely so this program has meaningful data to paginate through immediately — it checks if the table is empty and, if so, inserts 50 sample products. This runs once at startup in main.py, right after the tables are created.
  • skip: Annotated[int, Query(ge=0, ...)] and limit: Annotated[int, Query(ge=1, le=100, ...)] are the heart of this program:
    • Query(...) lets us attach validation constraints to query parameters, exactly the way Field(...) does for request bodies. ge=0 means “skip can’t be negative” — a negative offset makes no sense. le=100 on limit is an important safety cap: without it, a client could request limit=1000000 and defeat the entire purpose of pagination.
    • Default values (= 0, = 10) mean calling GET /products/ with no query parameters at all still works sensibly, returning the first 10 products.
  • get_paginated_products runs two queries: one to count the total number of matching rows (func.count()), and one to fetch just the requested slice (.offset(skip).limit(limit)). The total count is essential — without it, a client has no way to know how many pages exist or when they’ve reached the end.
  • PaginatedProducts wraps the actual items alongside total, skip, and limit — giving the client everything it needs to build page numbers or a “Showing 11–20 of 50” label, without needing a separate request.

Execution & Testing

uv run fastapi dev app/main.py
  1. Open http://127.0.0.1:8000/docs and call GET /products/ with no parameters — confirm you get the first 10 products and "total": 50.
  2. Call it again with skip=10&limit=10 — confirm you get products 11 through 20.
  3. Call it with skip=45&limit=10 — confirm you get only the final 5 products (46–50), since there aren’t 10 left.
  4. Try limit=500 — confirm FastAPI rejects it with a 422, since it exceeds the le=100 cap.
  5. Try skip=-5 — confirm FastAPI rejects it with a 422 as well, since it violates ge=0.

Program 2: Cursor-Based Pagination

Program Objective

Build a small, standalone FastAPI project with a GET /products/ endpoint using cursor-based pagination — returning a slice of data starting after a given product ID, along with a next_cursor the client uses to fetch the next batch, ideal for infinite-scroll style loading.

Step-by-Step Implementation

Step 1: Create the project

uv init cursor-pagination-demo
cd cursor-pagination-demo
uv add fastapi --extra standard
uv add pydantic-settings
uv add sqlalchemy

Step 2: Create the project structure

mkdir -p app/core app/db app/models app/schemas app/services app/routers
touch app/__init__.py app/core/__init__.py app/db/__init__.py app/models/__init__.py app/schemas/__init__.py app/services/__init__.py app/routers/__init__.py
cursor-pagination-demo/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── core/
│   │   ├── __init__.py
│   │   └── config.py
│   ├── db/
│   │   ├── __init__.py
│   │   └── database.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── product.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── product.py
│   ├── services/
│   │   ├── __init__.py
│   │   └── product_service.py
│   └── routers/
│       ├── __init__.py
│       └── products.py
├── pyproject.toml
└── uv.lock

Step 3: Create each file with the code below. Step 4: Run and test.

Complete Working Source Code

app/core/config.py

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    app_name: str = "Cursor Pagination Demo"
    app_version: str = "1.0.0"
    database_url: str = "sqlite:///./app.db"


settings = Settings()

app/db/database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

from app.core.config import settings

engine = create_engine(settings.database_url, connect_args={"check_same_thread": False})

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app/models/product.py

from sqlalchemy import Numeric, String
from sqlalchemy.orm import Mapped, mapped_column

from app.db.database import Base


class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(100), index=True)
    price: Mapped[float] = mapped_column(Numeric(10, 2))

app/schemas/product.py

from pydantic import BaseModel, ConfigDict


class ProductOut(BaseModel):
    id: int
    name: str
    price: float

    model_config = ConfigDict(from_attributes=True)


class CursorPaginatedProducts(BaseModel):
    items: list[ProductOut]
    next_cursor: int | None
    has_more: bool

app/services/product_service.py

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.models.product import Product


def seed_products_if_empty(db: Session) -> None:
    existing_count = db.scalar(select(func.count()).select_from(Product))
    if existing_count and existing_count > 0:
        return

    for i in range(1, 51):
        db.add(Product(name=f"Product {i:02d}", price=9.99 + i))
    db.commit()


def get_products_after_cursor(db: Session, cursor: int | None, limit: int) -> tuple[list[Product], bool]:
    query = db.query(Product).order_by(Product.id)

    if cursor is not None:
        query = query.filter(Product.id > cursor)

    # Fetch one extra row to determine if there's a next page, without a separate COUNT query.
    items = query.limit(limit + 1).all()

    has_more = len(items) > limit
    if has_more:
        items = items[:limit]

    return items, has_more

app/routers/products.py

from typing import Annotated

from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session

from app.db.database import get_db
from app.schemas.product import CursorPaginatedProducts
from app.services.product_service import get_products_after_cursor

router = APIRouter(prefix="/products", tags=["Products"])


@router.get("/", response_model=CursorPaginatedProducts)
def list_products(
    db: Session = Depends(get_db),
    cursor: Annotated[int | None, Query(description="ID to start after; omit for the first page")] = None,
    limit: Annotated[int, Query(ge=1, le=100, description="Max records to return")] = 10,
):
    items, has_more = get_products_after_cursor(db, cursor, limit)
    next_cursor = items[-1].id if items and has_more else None

    return CursorPaginatedProducts(items=items, next_cursor=next_cursor, has_more=has_more)

app/main.py

from fastapi import FastAPI

from app.core.config import settings
from app.db.database import Base, SessionLocal, engine
from app.models import product  # noqa: F401
from app.routers import products
from app.services.product_service import seed_products_if_empty

Base.metadata.create_all(bind=engine)

with SessionLocal() as db:
    seed_products_if_empty(db)

app = FastAPI(title=settings.app_name, version=settings.app_version)

app.include_router(products.router)


@app.get("/", tags=["Root"])
def read_root():
    return {"message": f"Welcome to {settings.app_name}"}

Detailed Code Explanation

  • cursor: int | None represents “the last ID the client already has.” On the very first request, the client omits it entirely (cursor=None), so no WHERE filter is applied and we start from the beginning.
  • query.filter(Product.id > cursor) is the core mechanic — instead of asking the database to skip past a number of rows (which gets progressively slower the deeper you page, since the database still has to walk through every skipped row), we ask it to jump straight to id > cursor using the primary key’s index. This stays fast regardless of how deep into the dataset you are.
  • The “fetch one extra row” trick (query.limit(limit + 1)) is a neat, efficient way to answer “is there a next page?” without running a separate COUNT query: if we asked for 10 and got 11 back, we know there’s more, so we trim the list back down to 10 and report has_more=True. If we got 10 or fewer, we’ve reached the end.
  • next_cursor = items[-1].id if items and has_more else None — the ID of the last item in the current batch becomes the cursor the client sends on their next request. When there’s no more data, next_cursor is None, signaling the client to stop requesting further pages.
  • Notice this endpoint doesn’t return a total count at all — that’s a deliberate trade-off of cursor-based pagination. Computing an exact total on a very large or frequently-changing table can itself be an expensive operation, and most infinite-scroll interfaces don’t actually need to show “50,000 of 2,000,000” — they just need to know whether to keep loading.

Execution & Testing

uv run fastapi dev app/main.py
  1. Call GET /products/ with no parameters — confirm you get the first 10 products, "has_more": true, and a next_cursor value (should be 10, the ID of the last item returned).
  2. Call GET /products/?cursor=10 — confirm you get products 11–20, along with a new next_cursor of 20.
  3. Keep following the next_cursor from each response as the next request’s cursor value — confirm you eventually reach a response with "has_more": false and "next_cursor": null, at product 50.
  4. Compare this to Program 1: notice there’s no way to jump directly to “page 3” here — you can only move forward one cursor at a time, which is the trade-off this pagination style makes in exchange for consistent performance at any depth.

Tips & Instructions for Developers

  • ✅ Always cap limit with le=... in Query(...) — an unbounded limit defeats the entire purpose of pagination.
  • ✅ Use offset/limit pagination for admin dashboards, search results, and anywhere users expect numbered pages they can jump between.
  • ✅ Use cursor-based pagination for feeds, activity logs, and any dataset that’s large, frequently changing, or loaded via infinite scroll.
  • ✅ Base your cursor on an indexed, sequential column (like a primary key or a created_at timestamp) — cursors only stay fast if the database can use an index to satisfy the WHERE clause.
  • ✅ Return enough metadata (total for offset pagination, has_more/next_cursor for cursor pagination) so the client never has to guess whether more data exists.
  • ❌ Don’t use plain OFFSET pagination on tables you expect to grow very large — the deeper a client pages, the slower each request gets, since the database still has to walk past every skipped row.
  • ❌ Don’t forget that data can change between paginated requests — a new row inserted between two offset-based page loads can cause an item to be skipped entirely or shown twice; cursor-based pagination is notably more resistant to this particular problem.

Official FastAPI Documentation References


Theory Questions

  1. Why does OFFSET-based pagination tend to get slower as the offset value grows larger?
  2. What problem does capping limit with le=100 protect your API against?
  3. Why does cursor-based pagination use WHERE id > cursor instead of counting and skipping rows?
  4. What’s the purpose of fetching limit + 1 rows in the cursor-based approach, instead of exactly limit?
  5. Why might an infinite-scroll feed choose not to expose a total count at all, even though it would be technically possible to compute?

Practical Assignment

  1. Add sorting options: Starting from Program 1’s project, add a sort_by query parameter (allowing "name" or "price") that changes the .order_by(...) clause used in get_paginated_products.
  2. Combine with search: Starting from Program 1’s project, add an optional search query parameter that filters products by a partial, case-insensitive name match (using .ilike()), applied before pagination, and make sure the total count reflects the filtered results, not the whole table.
  3. Simulate the “skipped row” problem: Starting from Program 1’s project, fetch skip=0&limit=10, then — using a separate request or a quick script — delete product with id=5 from the database, then fetch skip=10&limit=10. Compare the two responses and write down, in your own words, what went wrong and which product got silently skipped as a result.

Summary

In this chapter, you built two complete, runnable FastAPI projects covering both major pagination styles:

  • Classic skip/limit pagination with validated query parameters and full page metadata
  • Cursor-based pagination using an indexed WHERE id > cursor filter, well-suited for large or fast-growing datasets and infinite-scroll interfaces
Scroll to Top