Cursor pagination in FastAPI

Capsule overview

You already understand cursor pagination conceptually: base64 encoding, Page[T] with Pydantic, tuple comparison for the tiebreaker. Time to land it in real code.

This capsule is the end-to-end implementation. You're going to set up a FastAPI 0.110+ project with SQLAlchemy 2.0 async, write the GET /tasks endpoint that returns cursor pagination, handle every edge case (malformed cursor → HTTP 400, cursor for a different sort → HTTP 400, empty dataset, final page), and write tests with httpx.AsyncClient that validate the whole cycle.

By the end you'll have a working, runnable, testable endpoint that serves correct cursor pagination over PostgreSQL. It's the code you'll reuse in the module project (capsule 08) and in TaskFlow (module 8 of the guide).

Project setup

Dependencies

mkdir cursor-pagination-fastapi
cd cursor-pagination-fastapi
python -m venv venv
source venv/bin/activate

pip install \
  "fastapi[standard]>=0.110" \
  "sqlalchemy[asyncio]>=2.0" \
  "asyncpg>=0.29" \
  "pydantic>=2.6" \
  "uvicorn[standard]>=0.27" \
  "httpx>=0.27" \
  "pytest>=8.0" \
  "pytest-asyncio>=0.23"

Note: fastapi[standard] installs uvicorn and production-grade dependencies. SQLAlchemy 2.0+ is required for the select() syntax and modern async sessions.

File structure

cursor-pagination-fastapi/
├── app/
│   ├── __init__.py
│   ├── db.py                    # async session factory
│   ├── models.py                # SQLAlchemy ORM
│   ├── schemas.py               # Pydantic models (TaskOut, Page[T])
│   ├── pagination.py            # encode/decode + generic helpers
│   ├── repositories/
│   │   ├── __init__.py
│   │   └── tasks.py             # query with cursor
│   └── main.py                  # FastAPI app + endpoint
├── tests/
│   ├── conftest.py
│   └── test_pagination.py
└── seed.py                      # script to seed data

Database

# Create the DB
createdb cursor_demo

# Or with Docker
docker run -d --name pg-cursor-demo \
  -e POSTGRES_DB=cursor_demo \
  -e POSTGRES_PASSWORD=postgres \
  -p 5432:5432 \
  postgres:16

Model and schema

app/db.py

"""Configuration of the async connection to PostgreSQL."""
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/cursor_demo"

engine = create_async_engine(DATABASE_URL, echo=False, pool_pre_ping=True)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


async def get_session() -> AsyncSession:
    """FastAPI dependency: provides one async session per request."""
    async with AsyncSessionLocal() as session:
        yield session

app/models.py

"""ORM with SQLAlchemy 2.0 declarative_base."""
from datetime import datetime
from sqlalchemy import BigInteger, Index, String, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        server_default=func.now(),
    )

    # Composite index so ORDER BY (created_at DESC, id DESC) is a pure Index Scan.
    # Without this index, cursor pagination is still correct but NOT performant.
    __table_args__ = (
        Index(
            "idx_tasks_created_at_id_desc",
            created_at.desc(),
            id.desc(),
        ),
    )

Connection with guide #12: the composite index on (created_at DESC, id DESC) is what makes the plan use an Index Scan with no additional sorting. If your runtime sort doesn't match this index exactly, PostgreSQL can fall back to Sort + Seq Scan. The details of how to verify it with EXPLAIN ANALYZE are in guide #12 module 3.

app/schemas.py

"""Pydantic v2 schemas: request/response."""
from datetime import datetime
from typing import Generic, TypeVar
from pydantic import BaseModel, Field, ConfigDict

T = TypeVar("T")


class TaskOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    created_at: datetime


class Page(BaseModel, Generic[T]):
    """Generic paginated response with cursor pagination."""

    items: list[T] = Field(description="Items in this page, in sort order")
    next_cursor: str | None = Field(
        default=None,
        description="Cursor for the next page. None if there are no more items.",
    )
    has_more: bool = Field(description="True if there are additional items after this page")

Pagination helpers

app/pagination.py

"""Generic cursor pagination helpers: encode, decode, validation."""
import base64
import json
from datetime import datetime
from typing import Any

CURSOR_VERSION = 1


class CursorError(ValueError):
    """Malformed or incompatible cursor."""


def encode_cursor(payload: dict[str, Any]) -> str:
    """Encodes a dict as a URL-safe opaque cursor."""
    raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8")
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")


def decode_cursor(cursor: str) -> dict[str, Any]:
    """Decodes an opaque cursor. Raises CursorError if it's invalid."""
    try:
        padding = "=" * (-len(cursor) % 4)
        raw = base64.urlsafe_b64decode(cursor + padding)
        decoded = json.loads(raw)
    except (ValueError, json.JSONDecodeError) as e:
        raise CursorError("Malformed cursor") from e

    version = decoded.get("v")
    if version != CURSOR_VERSION:
        raise CursorError(
            f"Cursor version {version} not supported (current: v{CURSOR_VERSION})"
        )

    return decoded


def make_task_cursor(last_created_at: datetime, last_id: int) -> str:
    """Cursor specific to tasks ordered by (created_at DESC, id DESC)."""
    return encode_cursor({
        "v": CURSOR_VERSION,
        "t": last_created_at.isoformat().replace("+00:00", "Z"),
        "i": last_id,
        "d": "next",
    })


def parse_task_cursor(cursor: str) -> tuple[datetime, int]:
    """Parses the cursor into (created_at, id). Raises CursorError if it's wrong."""
    decoded = decode_cursor(cursor)
    try:
        ts = datetime.fromisoformat(decoded["t"].replace("Z", "+00:00"))
        last_id = int(decoded["i"])
    except (KeyError, ValueError, TypeError) as e:
        raise CursorError("Cursor with invalid fields") from e
    return ts, last_id

Repository: the query with a cursor

app/repositories/tasks.py

"""Data access for tasks: pagination with a cursor."""
from datetime import datetime
from sqlalchemy import select, tuple_
from sqlalchemy.ext.asyncio import AsyncSession

from app.models import Task
from app.pagination import (
    CursorError,
    make_task_cursor,
    parse_task_cursor,
)
from app.schemas import Page, TaskOut


async def list_tasks_paginated(
    session: AsyncSession,
    cursor: str | None = None,
    limit: int = 50,
) -> Page[TaskOut]:
    """
    Lists tasks paginated with a cursor.

    Args:
        session: SQLAlchemy AsyncSession.
        cursor: opaque cursor from the previous page (None = first page).
        limit: number of items per page.

    Returns:
        Page[TaskOut] with items, next_cursor, and has_more.

    Raises:
        CursorError: if the cursor is malformed or incompatible.
    """
    # 1. Start with the base query ordered by (created_at DESC, id DESC)
    stmt = select(Task).order_by(
        Task.created_at.desc(),
        Task.id.desc(),
    )

    # 2. If there's a cursor, apply the keyset filter with tuple comparison
    if cursor is not None:
        last_created_at, last_id = parse_task_cursor(cursor)
        # tuple_(...) generates SQL: WHERE (tasks.created_at, tasks.id) < (:t, :i)
        stmt = stmt.where(
            tuple_(Task.created_at, Task.id) < tuple_(last_created_at, last_id)
        )

    # 3. LIMIT + 1 to find out whether there's more without an extra query
    stmt = stmt.limit(limit + 1)

    # 4. Execute
    result = await session.execute(stmt)
    rows = result.scalars().all()

    # 5. Determine has_more and discard the extra row
    has_more = len(rows) > limit
    page_items = rows[:limit]

    # 6. Generate next_cursor from the last returned row
    next_cursor = None
    if has_more and page_items:
        last = page_items[-1]
        next_cursor = make_task_cursor(last.created_at, last.id)

    # 7. Convert ORM → Pydantic
    items_out = [TaskOut.model_validate(t) for t in page_items]

    return Page[TaskOut](
        items=items_out,
        next_cursor=next_cursor,
        has_more=has_more,
    )

Important details of the generated SQL:

-- The query SQLAlchemy generates for the first page (no cursor):
SELECT tasks.id, tasks.title, tasks.created_at
FROM tasks
ORDER BY tasks.created_at DESC, tasks.id DESC
LIMIT 51;

-- The query for subsequent pages (with a cursor):
SELECT tasks.id, tasks.title, tasks.created_at
FROM tasks
WHERE (tasks.created_at, tasks.id) < ($1, $2)
ORDER BY tasks.created_at DESC, tasks.id DESC
LIMIT 51;

The WHERE (tasks.created_at, tasks.id) < ($1, $2) is tuple comparison — PostgreSQL understands it natively. More in capsule 05.

FastAPI endpoint

app/main.py

"""FastAPI app: /tasks endpoint with cursor pagination."""
from fastapi import Depends, FastAPI, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import get_session
from app.pagination import CursorError
from app.repositories.tasks import list_tasks_paginated
from app.schemas import Page, TaskOut

app = FastAPI(title="Cursor Pagination Demo")


@app.get(
    "/tasks",
    response_model=Page[TaskOut],
    summary="List tasks with cursor pagination",
)
async def get_tasks(
    cursor: str | None = Query(
        default=None,
        description="Opaque cursor returned in next_cursor of the previous response.",
    ),
    limit: int = Query(
        default=50,
        ge=1,
        le=200,
        description="Number of items per page (max 200).",
    ),
    session: AsyncSession = Depends(get_session),
) -> Page[TaskOut]:
    """
    Lists tasks ordered by created_at DESC (most recent first).

    For the first page, don't send `cursor`.
    For subsequent pages, send the `next_cursor` from the previous response.
    """
    try:
        return await list_tasks_paginated(session, cursor=cursor, limit=limit)
    except CursorError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Invalid cursor: {e}",
        )

Notes:

• limit with ge=1, le=200 prevents the client from asking for a page of 1M items (a trivial DoS).
• CursorError is translated into an HTTP 400 — a client error, not a server one.
• response_model=Page[TaskOut] makes FastAPI validate and serialize the response automatically, and it exposes it correctly in /docs.

Seed: populating the DB with data

seed.py

"""Script to seed the DB with N random tasks."""
import asyncio
import random
from datetime import datetime, timedelta, timezone

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import AsyncSessionLocal, engine
from app.models import Base, Task


async def init_db() -> None:
    """Creates tables if they don't exist."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


async def seed(n: int) -> None:
    """Inserts n tasks with created_at spread across the last year."""
    now = datetime.now(timezone.utc)
    async with AsyncSessionLocal() as session:  # type: AsyncSession
        # Insert in batches of 5,000 so memory doesn't blow up
        batch_size = 5000
        for i in range(0, n, batch_size):
            batch = []
            for j in range(min(batch_size, n - i)):
                idx = i + j
                offset_seconds = random.randint(0, 365 * 24 * 3600)
                batch.append(
                    Task(
                        title=f"task_{idx}",
                        created_at=now - timedelta(seconds=offset_seconds),
                    )
                )
            session.add_all(batch)
            await session.commit()
            print(f"  inserted {i + len(batch)} / {n}")


async def main() -> None:
    await init_db()
    n = 10_000  # change to 1_000_000 for a serious benchmark
    print(f"Seeding {n} tasks...")
    await seed(n)
    print("Done.")


if __name__ == "__main__":
    asyncio.run(main())

python seed.py
# Seeding 10000 tasks...
#   inserted 5000 / 10000
#   inserted 10000 / 10000
# Done.

Bringing the endpoint up and testing it

uvicorn app.main:app --reload

# Page 1 (no cursor)
curl 'http://localhost:8000/tasks?limit=3' | jq

Expected output:

{
  "items": [
    {"id": 9847, "title": "task_9847", "created_at": "2026-04-30T18:23:12.456789+00:00"},
    {"id": 234,  "title": "task_234",  "created_at": "2026-04-30T17:55:01.123456+00:00"},
    {"id": 5621, "title": "task_5621", "created_at": "2026-04-30T16:42:30.789012+00:00"}
  ],
  "next_cursor": "eyJkIjoibmV4dCIsImkiOjU2MjEsInQiOiIyMDI2LTA0LTMwVDE2OjQyOjMwLjc4OTAxMloiLCJ2IjoxfQ",
  "has_more": true
}

# Page 2 (with page 1's cursor)
curl 'http://localhost:8000/tasks?limit=3&cursor=eyJkIjoibmV4dCIsImkiOjU2MjEsInQiOiIyMDI2LTA0LTMwVDE2OjQyOjMwLjc4OTAxMloiLCJ2IjoxfQ' | jq

{
  "items": [
    {"id": 1842, "title": "task_1842", "created_at": "2026-04-30T15:30:45.000123+00:00"},
    ...
  ],
  "next_cursor": "eyJkIjoibmV4dCIs...",
  "has_more": true
}

# Invalid cursor
curl -i 'http://localhost:8000/tasks?cursor=cursor_that_does_not_exist!!!'

HTTP/1.1 400 Bad Request
content-type: application/json

{"detail":"Invalid cursor: Malformed cursor"}

Tests with httpx.AsyncClient

tests/conftest.py

"""Async pytest fixtures for the endpoint tests."""
import asyncio
from typing import AsyncIterator

import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import AsyncSessionLocal, engine
from app.main import app
from app.models import Base, Task


@pytest_asyncio.fixture(scope="function")
async def setup_db() -> AsyncIterator[None]:
    """Creates a clean schema for each test."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)


@pytest_asyncio.fixture
async def session(setup_db) -> AsyncIterator[AsyncSession]:
    """AsyncSession for inserting test fixtures."""
    async with AsyncSessionLocal() as s:
        yield s


@pytest_asyncio.fixture
async def client(setup_db) -> AsyncIterator[AsyncClient]:
    """Async HTTP client to call the endpoint."""
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

tests/test_pagination.py

"""Tests of the /tasks endpoint with cursor pagination."""
from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.models import Task


pytestmark = pytest.mark.asyncio


async def _seed(session: AsyncSession, n: int) -> list[Task]:
    """Inserts n tasks with created_at spaced 1 minute apart. Returns the list in DESC order."""
    now = datetime.now(timezone.utc).replace(microsecond=0)
    tasks = [
        Task(title=f"task_{i}", created_at=now - timedelta(minutes=i))
        for i in range(n)
    ]
    session.add_all(tasks)
    await session.commit()
    # Refresh to get the autogenerated IDs
    for t in tasks:
        await session.refresh(t)
    return tasks


async def test_empty_page_without_cursor(client):
    """With no tasks in the DB, it returns empty items and has_more=False."""
    res = await client.get("/tasks")
    assert res.status_code == 200
    body = res.json()
    assert body["items"] == []
    assert body["next_cursor"] is None
    assert body["has_more"] is False


async def test_first_page_without_cursor(client, session):
    """With no cursor, it returns the first `limit` items."""
    await _seed(session, 5)

    res = await client.get("/tasks?limit=3")
    assert res.status_code == 200
    body = res.json()
    assert len(body["items"]) == 3
    assert body["has_more"] is True
    assert body["next_cursor"] is not None


async def test_navigate_all_pages(client, session):
    """Walking the pages sequentially returns every item, with no duplicates."""
    await _seed(session, 7)

    seen_ids = []
    cursor = None
    pages = 0

    while True:
        url = "/tasks?limit=3"
        if cursor:
            url += f"&cursor={cursor}"
        res = await client.get(url)
        assert res.status_code == 200
        body = res.json()

        seen_ids.extend(t["id"] for t in body["items"])
        cursor = body["next_cursor"]
        pages += 1

        if not body["has_more"]:
            break

        assert pages < 10  # safety net

    assert len(seen_ids) == 7  # saw them all
    assert len(set(seen_ids)) == 7  # no duplicates
    assert pages == 3  # 7 items / 3 per page = 3 pages (3+3+1)


async def test_invalid_cursor_returns_400(client):
    """A malformed cursor returns HTTP 400 with a clear message."""
    res = await client.get("/tasks?cursor=this_is_not_a_cursor!!")
    assert res.status_code == 400
    assert "Invalid cursor" in res.json()["detail"]


async def test_unknown_cursor_version_returns_400(client):
    """A cursor with an incompatible version returns 400."""
    import base64, json
    bad_payload = {"v": 999, "t": "2026-01-01T00:00:00Z", "i": 1}
    bad_cursor = base64.urlsafe_b64encode(
        json.dumps(bad_payload).encode("utf-8")
    ).rstrip(b"=").decode("ascii")

    res = await client.get(f"/tasks?cursor={bad_cursor}")
    assert res.status_code == 400
    assert "version 999" in res.json()["detail"]


async def test_limit_is_validated(client):
    """A limit out of range (0, >200) returns 422."""
    res = await client.get("/tasks?limit=0")
    assert res.status_code == 422

    res = await client.get("/tasks?limit=500")
    assert res.status_code == 422


async def test_order_is_created_at_desc(client, session):
    """Items come back ordered by created_at descending."""
    await _seed(session, 5)

    res = await client.get("/tasks?limit=10")
    body = res.json()
    timestamps = [t["created_at"] for t in body["items"]]
    assert timestamps == sorted(timestamps, reverse=True)


async def test_stable_under_recent_insertions(client, session):
    """
    Cursor pagination is stable under recent insertions:
    items inserted after the cursor don't appear in subsequent pages.
    """
    await _seed(session, 5)

    # Page 1: the two most recent items
    res = await client.get("/tasks?limit=2")
    page1 = res.json()
    page1_ids = [t["id"] for t in page1["items"]]

    # Insert a new task now (more recent than the seeded ones)
    new_task = Task(title="task_new", created_at=datetime.now(timezone.utc) + timedelta(hours=1))
    session.add(new_task)
    await session.commit()

    # Page 2 with page 1's cursor
    res = await client.get(f"/tasks?limit=2&cursor={page1['next_cursor']}")
    page2 = res.json()
    page2_ids = [t["id"] for t in page2["items"]]

    # The new task must NOT appear on page 2 (its created_at is more recent than the cursor)
    assert new_task.id not in page2_ids
    # And there are no duplicates between page 1 and page 2
    assert set(page1_ids).isdisjoint(page2_ids)

Running the tests

# Make sure the DB is running
pytest tests/ -v

Expected output:

tests/test_pagination.py::test_empty_page_without_cursor PASSED
tests/test_pagination.py::test_first_page_without_cursor PASSED
tests/test_pagination.py::test_navigate_all_pages PASSED
tests/test_pagination.py::test_invalid_cursor_returns_400 PASSED
tests/test_pagination.py::test_unknown_cursor_version_returns_400 PASSED
tests/test_pagination.py::test_limit_is_validated PASSED
tests/test_pagination.py::test_order_is_created_at_desc PASSED
tests/test_pagination.py::test_stable_under_recent_insertions PASSED

8 passed in 1.84s

Verify that the plan uses the index

Once you have the endpoint working, it's worth confirming that PostgreSQL is using the composite index (not a Seq Scan):

psql -d cursor_demo

EXPLAIN (ANALYZE, BUFFERS)
SELECT id, title, created_at FROM tasks
ORDER BY created_at DESC, id DESC
LIMIT 51;

Expected output:

Limit  (cost=0.42..2.83 rows=51 width=27) (actual time=0.018..0.205 rows=51 loops=1)
  Buffers: shared hit=4
  ->  Index Scan using idx_tasks_created_at_id_desc on tasks
        (cost=0.42..47204.42 rows=10000 width=27)
        (actual time=0.017..0.198 rows=51 loops=1)
        Buffers: shared hit=4
Planning Time: 0.090 ms
Execution Time: 0.230 ms

Verify:

• ✅ Index Scan using idx_tasks_created_at_id_desc (not a Seq Scan)
• ✅ Buffers: shared hit=4 (it reads only what's needed)
• ✅ Execution Time: 0.230 ms (constant)

Now with a cursor:

EXPLAIN (ANALYZE, BUFFERS)
SELECT id, title, created_at FROM tasks
WHERE (created_at, id) < ('2026-04-15 10:23:45+00'::timestamptz, 12345)
ORDER BY created_at DESC, id DESC
LIMIT 51;

Expected output:

Limit  (cost=0.42..2.83 rows=51 width=27) (actual time=0.025..0.215 rows=51 loops=1)
  Buffers: shared hit=4
  ->  Index Scan using idx_tasks_created_at_id_desc on tasks
        (cost=0.42..23502.21 rows=5000 width=27)
        (actual time=0.024..0.208 rows=51 loops=1)
        Index Cond: (ROW(created_at, id) < ROW('2026-04-15 10:23:45+00'::timestamptz, 12345))
        Buffers: shared hit=4
Planning Time: 0.108 ms
Execution Time: 0.245 ms

The critical part: Index Cond: (ROW(created_at, id) < ROW(...)) — PostgreSQL applied the filter inside the Index Scan, not as a post-scan filter. That's what makes cursor pagination O(log n).

Deeper coverage of EXPLAIN ANALYZE and how to tell Index Cond apart from Filter is in guide #12 module 2. This capsule assumes you already understand the difference.

Why does this matter in real work?

1. A production-ready endpoint. What you wrote is code that would pass code review on any serious team: types, input validation, error handling with semantically correct HTTP, tests for the happy path and edge cases. It isn't "tutorial code" — it's real code.

2. A reusable pattern. The repositories/tasks.py + pagination.py + schemas.py structure replicates identically for any other entity. When tomorrow you need cursor pagination on /events or /messages, copy-paste with renames and you're done.

3. Input validation with Pydantic. Query(default=50, ge=1, le=200) is the first line of defense against abuse. Without it, someone can ask for ?limit=999999999 and knock over your DB. Framework-level validation protects you before the query is even built.

4. Tests with httpx.AsyncClient. The client + setup_db test pattern is the standard in serious FastAPI projects. It lets you test the full endpoint (including Pydantic serialization, dependency injection, error handling) without bringing up uvicorn — all in-process, fast, reproducible.

Traps and common mistakes

Mistake 1 (conceptual): not including the composite index

Symptom: the endpoint works, but on the first page with a large table, EXPLAIN shows a Seq Scan or an Index Scan + Sort.

Why it happens: without idx_tasks_created_at_id_desc, PostgreSQL has no efficient way to return rows in (created_at DESC, id DESC) order. The behavior "works" but it's slow.

How to tell: run the EXPLAIN ANALYZE we showed above. If you see Sort in the plan, you're missing the index. If you see a direct Index Scan, you're fine.

How to fix it: make sure the index was created:

\d tasks
-- Should show:
-- Indexes:
--   "tasks_pkey" PRIMARY KEY, btree (id)
--   "idx_tasks_created_at_id_desc" btree (created_at DESC, id DESC)

Mistake 2 (practical): mixing selectinload with a cursor without minding the order

Symptom: you add eager loading (selectinload(Task.author)) and pagination stops working properly.

Why it happens: selectinload runs a second query (SELECT * FROM authors WHERE id IN (...)) that does NOT follow the first one's order. But the first query's ORDER BY is still intact, so the items come back ordered and SQLAlchemy does the join in memory. It generally works — but if you combine it with joinedload (LEFT OUTER JOIN), the LIMIT can behave strangely because it counts rows from the JOIN, not from the entity.

How to fix it:

• For cursor pagination, prefer selectinload over joinedload when there are one-to-many relationships.
• If you need joinedload, use a subquery: paginate the main entity in a subquery, then apply joinedload on the result.

Mistake 3 (conceptual): using > instead of < in the WHERE

Symptom: pagination returns items in inverted order or "gets stuck on the first page."

Why it happens: confusing the operator. If the sort is DESC, you want "older" items after the cursor → <. If the sort is ASC, you want "newer" items → >.

How to tell: look at the generated SQL. For ORDER BY created_at DESC, id DESC with a cursor, the filter must be <:

WHERE (tasks.created_at, tasks.id) < ($1, $2)  -- Sort DESC: filter <

If it were ASC:

WHERE (tasks.created_at, tasks.id) > ($1, $2)  -- Sort ASC: filter >

Mistake 4 (edge case): a last-page cursor that returns more

Symptom: you reach the last page (has_more=false), but someone inserted items in the meantime. The client keeps sending the previous cursor and starts receiving new items as if it were the "next page."

Why it happens: cursor is stable under recent insertions (new items don't wedge themselves between pages you already saw). But if you reach the "end" and then items get inserted with created_at < the cursor, those items show up in the next call with the old cursor.

How to tell: it's expected behavior, not a bug. But if the UI assumes "when has_more=false, there's nothing more forever," the client can get confused.

How to fix it:

• Document that has_more=false means "there's no more right now," not "there will never be more."
• If the UX requires a "definitive end," the client should discard the cursor when has_more=false and not call again.

Mistake 5 (practical): missing from_attributes=True in Pydantic v2

Symptom: TaskOut.model_validate(task_orm_object) fails with the error "expected dict, got Task".

Why it happens: Pydantic v2 doesn't assume an ORM object can be converted straight into a dict. You need model_config = ConfigDict(from_attributes=True) (the equivalent of v1's orm_mode = True).

How to fix it: it's already in the example's app/schemas.py. If you see this error, check that your Pydantic model has that config.

Exercises

Exercise 1: add logging of the generated SQL

Enable echo=True on the engine in app/db.py and observe the SQL SQLAlchemy generates for page 1 vs page N. Confirm that the WHERE clause with a cursor is exactly (tasks.created_at, tasks.id) < (...).

# app/db.py
engine = create_async_engine(DATABASE_URL, echo=True, pool_pre_ping=True)

uvicorn app.main:app --reload

curl 'http://localhost:8000/tasks?limit=2'

INFO  sqlalchemy.engine.Engine BEGIN
INFO  sqlalchemy.engine.Engine SELECT tasks.id, tasks.title, tasks.created_at
FROM tasks ORDER BY tasks.created_at DESC, tasks.id DESC
 LIMIT $1::INTEGER
INFO  sqlalchemy.engine.Engine [generated in 0.00040s] (3,)

curl 'http://localhost:8000/tasks?limit=2&cursor=eyJ...'

INFO  sqlalchemy.engine.Engine SELECT tasks.id, tasks.title, tasks.created_at
FROM tasks
WHERE (tasks.created_at, tasks.id) < ($1::TIMESTAMP WITH TIME ZONE, $2::BIGINT)
ORDER BY tasks.created_at DESC, tasks.id DESC
 LIMIT $3::INTEGER
INFO  sqlalchemy.engine.Engine [...] (datetime.datetime(2026, 4, 15, 10, 23, 45, ...), 12345, 3)

Exercise 2: add previous_cursor for simple bidirectional navigation

Your current Page[T] only has next_cursor. Add a previous_cursor that points to the first item of the current page (what the client would use with direction=prev). For now, just add the field and the cursor — the full backward-navigation logic comes in capsule 06.

# app/schemas.py
class Page(BaseModel, Generic[T]):
    items: list[T]
    next_cursor: str | None = None
    previous_cursor: str | None = None  # NEW
    has_more: bool

# app/repositories/tasks.py
async def list_tasks_paginated(
    session: AsyncSession,
    cursor: str | None = None,
    limit: int = 50,
) -> Page[TaskOut]:
    # ... previous logic ...

    # Generate previous_cursor from the FIRST returned item
    previous_cursor = None
    if page_items and cursor is not None:
        # It only makes sense to return previous if we came from an earlier page
        first = page_items[0]
        previous_cursor = make_task_cursor(first.created_at, first.id)
        # Note: this cursor will represent "items with created_at >= the first one"
        # The query with direction='prev' is what you implement in capsule 06.

    return Page[TaskOut](
        items=items_out,
        next_cursor=next_cursor,
        previous_cursor=previous_cursor,
        has_more=has_more,
    )

curl 'http://localhost:8000/tasks?limit=2'
# {"items": [...], "next_cursor": "...", "previous_cursor": null, "has_more": true}

curl 'http://localhost:8000/tasks?limit=2&cursor=eyJ...'
# {"items": [...], "next_cursor": "...", "previous_cursor": "eyJ...", "has_more": true}

Exercise 3: rate-limit queries per client (cursor-dependent)

Implement a simple middleware that rate-limits by IP only when more than 10 sequential pages are requested in under 60 seconds. Not counting individual pages — just the "aggressive pagination loop" pattern.

# app/middleware/rate_limit.py
from collections import defaultdict, deque
from time import monotonic
from fastapi import Request, HTTPException, status

# In-memory tracker (in production, use Redis)
WINDOW_SECONDS = 60
MAX_REQUESTS_WITH_CURSOR = 10
_history: dict[str, deque[float]] = defaultdict(deque)


async def rate_limit_pagination(request: Request, call_next):
    # Only apply to /tasks with a cursor (a pagination loop)
    is_pagination = (
        request.url.path == "/tasks"
        and request.query_params.get("cursor") is not None
    )

    if is_pagination:
        ip = request.client.host
        now = monotonic()
        # Clear requests outside the window
        history = _history[ip]
        while history and history[0] < now - WINDOW_SECONDS:
            history.popleft()

        if len(history) >= MAX_REQUESTS_WITH_CURSOR:
            return JSONResponse(
                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                content={"detail": "Too many pagination requests, slow down."},
            )
        history.append(now)

    return await call_next(request)


# app/main.py
from app.middleware.rate_limit import rate_limit_pagination
app.middleware("http")(rate_limit_pagination)

# Make 11 calls with a cursor in under 60s
for i in $(seq 1 11); do
  curl -s -o /dev/null -w "%{http_code}\n" \
    'http://localhost:8000/tasks?cursor=eyJ...'
done

# Output: 200 200 200 200 200 200 200 200 200 200 429

Exercise 4: add OpenAPI examples to the endpoint

Modify the /tasks endpoint so /docs shows concrete request and response examples. Useful so consumers can see what to expect.

# app/main.py
from fastapi import Depends, FastAPI, HTTPException, Query, status

@app.get(
    "/tasks",
    response_model=Page[TaskOut],
    summary="List tasks with cursor pagination",
    responses={
        200: {
            "description": "A page of tasks with a cursor for the next one",
            "content": {
                "application/json": {
                    "examples": {
                        "first_page": {
                            "summary": "First page (no cursor)",
                            "value": {
                                "items": [
                                    {"id": 12345, "title": "Review PR", "created_at": "2026-04-15T10:23:45.123Z"},
                                    {"id": 12344, "title": "Update docs", "created_at": "2026-04-15T10:22:30.000Z"},
                                ],
                                "next_cursor": "eyJ0IjoiMjAyNi0w...",
                                "has_more": True,
                            },
                        },
                        "last_page": {
                            "summary": "Last page",
                            "value": {
                                "items": [{"id": 1, "title": "First task", "created_at": "2026-01-01T00:00:00Z"}],
                                "next_cursor": None,
                                "has_more": False,
                            },
                        },
                    }
                }
            },
        },
        400: {
            "description": "Invalid cursor",
            "content": {
                "application/json": {
                    "example": {"detail": "Invalid cursor: Malformed cursor"}
                }
            },
        },
    },
)
async def get_tasks(...):
    ...

uvicorn app.main:app --reload
# Open http://localhost:8000/docs
# The /tasks endpoint now shows clickable examples.

Summary and next step

In this capsule you built:

• A working FastAPI endpoint with cursor pagination using SQLAlchemy 2.0 async + asyncpg + PostgreSQL 16+.
• A composite index on (created_at DESC, id DESC) that enables a pure Index Scan (verified with EXPLAIN ANALYZE).
• SQLAlchemy's tuple_(), which translates directly into PostgreSQL tuple comparison — the only correct way to paginate with a tiebreaker.
• Error handling with semantically correct HTTP: invalid cursor → 400, limit out of range → 422, no items → 200 with an empty array.
• Tests with httpx.AsyncClient that validate the happy path, edge cases, and stability under insertions.
• Input validation with Pydantic that protects against abuse (limit bounds, cursor validation).

Before moving on you should be able to:

• Implement cursor pagination for a new entity (e.g. events, messages) in under 30 minutes
• Verify with EXPLAIN ANALYZE that your query uses an Index Scan with tuple comparison
• Write tests for the happy path and at least 3 edge cases (invalid cursor, empty page, walking every page)
• Tell the correct tuple_() apart from the wrong form (comparing column by column with AND/OR)

Next capsule — Keyset pagination and when to use it. You're going to go deeper into the SQL behind the cursor: composite cursors with tuple comparison, cases where pure keyset (without opacity) beats an opaque cursor, sorting by mixed columns (some ASC, some DESC), and why the WHERE (a, b) < ($1, $2) syntax is the only one the planner optimizes correctly — versus the trap of WHERE a < $1 OR (a = $1 AND b < $2), which looks equivalent but produces worse plans.

Resources

1. FastAPI — Pagination patterns — FastAPI's official reference. Although it doesn't have a page specifically on cursor pagination, the Query parameter and dependency injection patterns we use are documented here.
2. SQLAlchemy 2.0 — tuple_() operator — the official documentation of the helper that generates tuple comparison.
3. SQLAlchemy 2.0 — Async Quickstart — a reference for async sessions, the engine, and dependency injection.
4. asyncpg documentation — the fastest async PostgreSQL driver in Python. We use it as SQLAlchemy's backend.
5. Pydantic v2 — Generic models — the reference for the generic Page[T].
6. Markus Winand — "Pagination Done the PostgreSQL Way" — a talk with real PostgreSQL code, complementing what we saw.
7. Brandur Leach — "Building Robust APIs" — good error handling and validation patterns that we applied in the endpoint.
8. Real World FastAPI — pagination example (GitHub) — a reference project that uses similar patterns.

Module 1 — SQL Patterns for Production APIs Guide

Next capsule: Keyset pagination and when to use it — the SQL behind the cursor, composite cursors, and tuple comparison.