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/limitpagination 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 / Limit | Cursor-Based | |
|---|---|---|
| How it works | “Skip the first N rows, then give me the next M” | “Give me M rows after this specific position” |
| Client sends | skip=20&limit=10 | cursor=87&limit=10 |
| Good for | Traditional numbered pages (“Page 1, 2, 3…”) | Infinite scroll, very large or frequently-changing tables |
| Downside | Gets slower as skip grows — the database still has to count past all those skipped rows | Can’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_emptyexists 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 inmain.py, right after the tables are created.skip: Annotated[int, Query(ge=0, ...)]andlimit: Annotated[int, Query(ge=1, le=100, ...)]are the heart of this program:Query(...)lets us attach validation constraints to query parameters, exactly the wayField(...)does for request bodies.ge=0means “skip can’t be negative” — a negative offset makes no sense.le=100onlimitis an important safety cap: without it, a client could requestlimit=1000000and defeat the entire purpose of pagination.- Default values (
= 0,= 10) mean callingGET /products/with no query parameters at all still works sensibly, returning the first 10 products.
get_paginated_productsruns 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.PaginatedProductswraps the actualitemsalongsidetotal,skip, andlimit— 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
- Open
http://127.0.0.1:8000/docsand callGET /products/with no parameters — confirm you get the first 10 products and"total": 50. - Call it again with
skip=10&limit=10— confirm you get products 11 through 20. - Call it with
skip=45&limit=10— confirm you get only the final 5 products (46–50), since there aren’t 10 left. - Try
limit=500— confirm FastAPI rejects it with a422, since it exceeds thele=100cap. - Try
skip=-5— confirm FastAPI rejects it with a422as well, since it violatesge=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 | Nonerepresents “the last ID the client already has.” On the very first request, the client omits it entirely (cursor=None), so noWHEREfilter 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 toid > cursorusing 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 separateCOUNTquery: if we asked for 10 and got 11 back, we know there’s more, so we trim the list back down to 10 and reporthas_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_cursorisNone, signaling the client to stop requesting further pages.- Notice this endpoint doesn’t return a
totalcount 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
- Call
GET /products/with no parameters — confirm you get the first 10 products,"has_more": true, and anext_cursorvalue (should be10, the ID of the last item returned). - Call
GET /products/?cursor=10— confirm you get products 11–20, along with a newnext_cursorof20. - Keep following the
next_cursorfrom each response as the next request’scursorvalue — confirm you eventually reach a response with"has_more": falseand"next_cursor": null, at product 50. - 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
limitwithle=...inQuery(...)— 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_attimestamp) — cursors only stay fast if the database can use an index to satisfy theWHEREclause. - ✅ Return enough metadata (
totalfor offset pagination,has_more/next_cursorfor cursor pagination) so the client never has to guess whether more data exists. - ❌ Don’t use plain
OFFSETpagination 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
- Query Parameters and String Validations
- Query Parameter Models
- SQLAlchemy — Selecting Rows with Core or ORM
- SQLAlchemy — LIMIT/OFFSET
Theory Questions
- Why does
OFFSET-based pagination tend to get slower as the offset value grows larger? - What problem does capping
limitwithle=100protect your API against? - Why does cursor-based pagination use
WHERE id > cursorinstead of counting and skipping rows? - What’s the purpose of fetching
limit + 1rows in the cursor-based approach, instead of exactlylimit? - Why might an infinite-scroll feed choose not to expose a
totalcount at all, even though it would be technically possible to compute?
Practical Assignment
- Add sorting options: Starting from Program 1’s project, add a
sort_byquery parameter (allowing"name"or"price") that changes the.order_by(...)clause used inget_paginated_products. - Combine with search: Starting from Program 1’s project, add an optional
searchquery parameter that filters products by a partial, case-insensitive name match (using.ilike()), applied before pagination, and make sure thetotalcount reflects the filtered results, not the whole table. - 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 withid=5from the database, then fetchskip=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/limitpagination with validated query parameters and full page metadata - Cursor-based pagination using an indexed
WHERE id > cursorfilter, well-suited for large or fast-growing datasets and infinite-scroll interfaces
