Pagination with filters and dynamic ordering

Capsule overview

Up to here your endpoint is GET /tasks?cursor=...&limit=.... It works perfectly for the demo, but the reality is that real APIs look more like this:

GET /tasks?status=active&priority=high&assignee_id=42&sort=updated_at&order=desc&cursor=...&limit=50

Four filters + dynamic sort + cursor + limit. This capsule shows you how to combine them without breaking the cursor's opacity, without allowing SQL injection in the sort, and with the right balance between flexibility for the client and simplicity for you.

You're going to learn:

• How filters combine with cursor pagination. Spoiler: filters go outside the cursor, not inside.
• Why changing filters invalidates the cursor. And how to handle that case explicitly.
• How to allow dynamic ordering without SQL injection. Allowlist + enum, never string concatenation.
• When a dynamic sort requires new indexes (and when it doesn't).
• The "flexible API vs maintainable API" trade-off: what looks like a power-user feature ends up being technical debt.

Rule #1: filters go outside the cursor

When you add filters (status=active), the instinct is "I'll put them in the cursor so they persist between pages." It's tempting and it's wrong.

# ❌ ANTI-PATTERN: filters inside the cursor
{
    "v": 1,
    "t": "2026-04-15T10:23:45Z",
    "i": 12345,
    "filters": {"status": "active", "priority": "high"}  # NO
}

Why it's an anti-pattern:

1. It couples the cursor to the filters. If the client changes a filter between pages, the "old" cursor keeps carrying the old filters — confusing behavior.
2. It bloats the cursor. Every filter adds bytes. Five filters and your cursor is 200+ characters.
3. It leaks the internal schema. The client can read which filters you accept (even with HMAC).
4. It breaks REST conventions. Filters belong in query params, not hidden away.

The correct pattern: filters as separate query params

Page 1:
GET /tasks?status=active&priority=high&limit=50
  ↓
{"items": [...], "next_cursor": "eyJ0Ijoi..."}

Page 2 (the client sends the filters + cursor back):
GET /tasks?status=active&priority=high&cursor=eyJ0Ijoi...&limit=50

Rules:

• The cursor only carries position info (timestamp, id, direction, version, key_id).
• Filters go in query params, repeated in every request.
• The client is responsible for keeping the filters consistent between pages.

And what if the client changes a filter between pages? Your API treats every request as independent. If page 2 comes in with ?status=closed&cursor=..., you return items with status=closed that come after the cursor. It's predictable behavior.

Worked example: an endpoint with filters

We're going to extend the endpoint from capsule 04 to support filters on status and priority.

Updated model

# app/models.py
from sqlalchemy import BigInteger, Index, String, DateTime, Integer, Enum as SAEnum, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime
import enum


class TaskStatus(str, enum.Enum):
    open = "open"
    in_progress = "in_progress"
    closed = "closed"


class TaskPriority(str, enum.Enum):
    low = "low"
    medium = "medium"
    high = "high"


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)
    status: Mapped[TaskStatus] = mapped_column(
        SAEnum(TaskStatus, name="task_status"),
        nullable=False,
        default=TaskStatus.open,
    )
    priority: Mapped[TaskPriority] = mapped_column(
        SAEnum(TaskPriority, name="task_priority"),
        nullable=False,
        default=TaskPriority.medium,
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )

    __table_args__ = (
        # Composite index for sorting by created_at (the default)
        Index("idx_tasks_created_id_desc", created_at.desc(), id.desc()),
        # Composite indexes that match the most common filters:
        # status filter + created_at sort:
        Index(
            "idx_tasks_status_created_id",
            status,
            created_at.desc(),
            id.desc(),
        ),
        # priority filter + created_at sort:
        Index(
            "idx_tasks_priority_created_id",
            priority,
            created_at.desc(),
            id.desc(),
        ),
    )

Deeper coverage of how to choose composite indexes when there are multiple filters (a common case: one per filter, or one multi-filter index?) is in guide #12 module 3. For this capsule, a pragmatic rule: a composite index per filter + the sort column(s), in that order.

Schema with filters

# app/schemas.py
from pydantic import BaseModel, ConfigDict, Field
from datetime import datetime
from typing import Generic, TypeVar
import enum


class TaskStatusOut(str, enum.Enum):
    open = "open"
    in_progress = "in_progress"
    closed = "closed"


class TaskPriorityOut(str, enum.Enum):
    low = "low"
    medium = "medium"
    high = "high"


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

    id: int
    title: str
    status: TaskStatusOut
    priority: TaskPriorityOut
    created_at: datetime


T = TypeVar("T")


class Page(BaseModel, Generic[T]):
    items: list[T]
    next_cursor: str | None = None
    previous_cursor: str | None = None
    has_more: bool

Repository with filters

# app/repositories/tasks.py
from sqlalchemy import select, tuple_, ColumnElement
from sqlalchemy.ext.asyncio import AsyncSession

from app.models import Task, TaskStatus, TaskPriority
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,
    status: TaskStatus | None = None,
    priority: TaskPriority | None = None,
) -> Page[TaskOut]:
    # 1. Base ordered query
    stmt = select(Task).order_by(Task.created_at.desc(), Task.id.desc())

    # 2. Apply the filters (they go OUTSIDE the cursor)
    if status is not None:
        stmt = stmt.where(Task.status == status)
    if priority is not None:
        stmt = stmt.where(Task.priority == priority)

    # 3. Apply the cursor (position only)
    direction = "next"
    if cursor is not None:
        cursor_t, cursor_i, direction = parse_task_cursor(cursor)
        if direction == "next":
            stmt = stmt.where(
                tuple_(Task.created_at, Task.id) < tuple_(cursor_t, cursor_i)
            )
        else:
            # direction == "prev"
            stmt = stmt.where(
                tuple_(Task.created_at, Task.id) > tuple_(cursor_t, cursor_i)
            ).order_by(None).order_by(Task.created_at.asc(), Task.id.asc())

    stmt = stmt.limit(limit + 1)
    result = await session.execute(stmt)
    rows = list(result.scalars().all())

    has_more = len(rows) > limit
    page_items = rows[:limit]

    if direction == "prev":
        page_items.reverse()

    next_cursor = None
    previous_cursor = None
    if page_items:
        first, last = page_items[0], page_items[-1]
        if direction == "next" and has_more:
            next_cursor = make_task_cursor(last.created_at, last.id, "next")
        elif direction == "prev":
            next_cursor = make_task_cursor(last.created_at, last.id, "next")
        if direction == "next" and cursor is not None:
            previous_cursor = make_task_cursor(first.created_at, first.id, "prev")
        elif direction == "prev" and has_more:
            previous_cursor = make_task_cursor(first.created_at, first.id, "prev")

    return Page[TaskOut](
        items=[TaskOut.model_validate(t) for t in page_items],
        next_cursor=next_cursor,
        previous_cursor=previous_cursor,
        has_more=has_more,
    )

Endpoint with filters

# app/main.py
from fastapi import Depends, FastAPI, HTTPException, Query, status as http_status
from app.models import TaskStatus, TaskPriority


@app.get("/tasks", response_model=Page[TaskOut])
async def get_tasks(
    cursor: str | None = Query(default=None),
    limit: int = Query(default=50, ge=1, le=200),
    status: TaskStatus | None = Query(default=None, description="Filter by status"),
    priority: TaskPriority | None = Query(
        default=None, description="Filter by priority"
    ),
    session: AsyncSession = Depends(get_session),
) -> Page[TaskOut]:
    try:
        return await list_tasks_paginated(
            session,
            cursor=cursor,
            limit=limit,
            status=status,
            priority=priority,
        )
    except CursorError as e:
        raise HTTPException(
            status_code=http_status.HTTP_400_BAD_REQUEST,
            detail=f"Invalid cursor: {e}",
        )

FastAPI automatically validates that status is one of open, in_progress, closed because you used TaskStatus as the type. If the client sends ?status=invalid, FastAPI returns HTTP 422 before your code ever runs.

Trying it out

# No filters
curl 'http://localhost:8000/tasks?limit=3' | jq '.items | length'
# 3

# Filter by status
curl 'http://localhost:8000/tasks?status=open&limit=3' | jq
# {"items": [...only open...], ...}

# Filter by status + priority
curl 'http://localhost:8000/tasks?status=open&priority=high&limit=3' | jq

# Filters + cursor (second page)
curl 'http://localhost:8000/tasks?status=open&priority=high&cursor=eyJ...' | jq

# Invalid filter (FastAPI rejects it)
curl 'http://localhost:8000/tasks?status=invalid' | jq
# {"detail": [{"loc": [...], "msg": "Input should be 'open', 'in_progress' or 'closed'", ...}]}

Dynamic ordering: the dangerous case

Up to here the sort is fixed: ORDER BY created_at DESC, id DESC. Some clients want to choose:

GET /tasks?sort=priority         # order by priority DESC
GET /tasks?sort=updated_at       # order by updated_at DESC
GET /tasks?sort=title            # order by title ASC

There are two ways to implement it. One is correct, the other is vulnerable to SQL injection.

❌ The wrong way (SQL injection)

# NEVER DO THIS
async def get_tasks(sort: str = "created_at"):
    query = f"SELECT * FROM tasks ORDER BY {sort} DESC LIMIT 50"
    # If sort = "id; DROP TABLE tasks; --", catastrophe.
    result = await session.execute(text(query))

Even if you use SQLAlchemy:

# ALSO VULNERABLE
stmt = select(Task).order_by(text(f"{sort} DESC")).limit(50)
# text() doesn't parameterize identifiers, only values

SQL injection with identifiers (column and table names) is real. PostgreSQL doesn't parameterize identifiers, only values. The only defense is not to concatenate.

✅ The correct way: allowlist + Enum

# app/schemas.py
class TaskSortField(str, enum.Enum):
    created_at = "created_at"
    updated_at = "updated_at"
    priority = "priority"
    title = "title"


class SortOrder(str, enum.Enum):
    asc = "asc"
    desc = "desc"

# app/repositories/tasks.py
from sqlalchemy import asc, desc

# Explicit mapping of enum → SQLAlchemy column
SORT_FIELDS_MAP = {
    TaskSortField.created_at: Task.created_at,
    TaskSortField.updated_at: Task.updated_at,
    TaskSortField.priority: Task.priority,
    TaskSortField.title: Task.title,
}


def get_sort_columns(
    sort: TaskSortField,
    order: SortOrder,
) -> list[ColumnElement]:
    """Builds the ORDER BY safely using an allowlist."""
    sort_col = SORT_FIELDS_MAP[sort]  # KeyError if it isn't there, which is what we want
    direction = desc if order == SortOrder.desc else asc

    # Always include id as a tiebreaker
    return [direction(sort_col), direction(Task.id)]

# Endpoint
@app.get("/tasks")
async def get_tasks(
    sort: TaskSortField = Query(default=TaskSortField.created_at),
    order: SortOrder = Query(default=SortOrder.desc),
    ...
):
    ...

FastAPI automatically rejects values not in the Enum. ?sort=evil_string returns 422 before touching the query.

The Enum is the allowlist. Even if someone found a way to bypass FastAPI (extremely unlikely), SORT_FIELDS_MAP[sort] raises KeyError for any unknown value. Double defense.

Cursor + dynamic sort: the subtle problem

Here comes the detail almost every tutorial skips. If the sort is dynamic, the cursor depends on the sort.

Sort by created_at: cursor carries (created_at, id) → ORDER BY (created_at, id) DESC
Sort by priority:   cursor carries (priority, id)   → ORDER BY (priority, id) DESC
Sort by updated_at: cursor carries (updated_at, id) → ORDER BY (updated_at, id) DESC

If the client changes the sort between pages, the previous cursor is invalid.

Approach 1: the cursor includes the sort

{
    "v": 1,
    "s": "priority",                # sort field
    "o": "desc",                    # order
    "k": [5, 12345],                # keys: [priority_value, id]
    "d": "next"
}

Validation: if the client sends ?sort=created_at&cursor=... but the cursor says s=priority, you reject with HTTP 400.

def parse_dynamic_cursor(cursor: str, expected_sort: TaskSortField) -> tuple:
    decoded = decode_signed_cursor(cursor)
    if decoded.get("s") != expected_sort.value:
        raise CursorError(
            f"Cursor belongs to sort='{decoded.get('s')}', "
            f"but the request asks for sort='{expected_sort.value}'"
        )
    return decoded["k"], decoded.get("d", "next")

Pros: explicit; the client knows which cursor belongs to which sort. Cons: 400 errors when the client changes the sort. The UX requires the client to restart pagination.

Approach 2: the client restarts pagination when changing the sort

Simpler:

• When the client changes the sort, it discards the cursor and starts from page 1.
• Your API doesn't need to validate that the cursor belongs to the current sort.

Pros: simpler code, fewer cases to handle. Cons: the UX requires restarting pagination when changing the sort. Most UIs do this naturally (a sort change = scroll to top).

Recommendation

For this module, approach 1. It's what TaskFlow (module 8) and most serious APIs implement. The validation is 5 lines of code and it prevents an entire class of bug (a cursor from a different sort causing strange results).

Implementation with a dynamic sort

# app/pagination_dynamic.py
from typing import Literal


def make_dynamic_cursor(
    sort_field: TaskSortField,
    order: SortOrder,
    keys: tuple,  # (sort_value, id)
    direction: Literal["next", "prev"],
) -> str:
    return encode_signed_cursor({
        "v": 1,
        "s": sort_field.value,
        "o": order.value,
        "k": list(keys),
        "d": direction,
    })


def parse_dynamic_cursor(
    cursor: str,
    expected_sort: TaskSortField,
    expected_order: SortOrder,
) -> tuple[list, str]:
    decoded = decode_signed_cursor(cursor)

    if decoded.get("s") != expected_sort.value:
        raise CursorError(
            f"Cursor belongs to sort='{decoded.get('s')}', request uses sort='{expected_sort.value}'"
        )
    if decoded.get("o") != expected_order.value:
        raise CursorError(
            f"Cursor belongs to order='{decoded.get('o')}', request uses order='{expected_order.value}'"
        )

    return decoded["k"], decoded.get("d", "next")

And the query uses the keys correctly:

async def list_tasks_dynamic_sort(
    session: AsyncSession,
    sort: TaskSortField,
    order: SortOrder,
    cursor: str | None,
    limit: int,
    status: TaskStatus | None = None,
):
    sort_col = SORT_FIELDS_MAP[sort]

    stmt = select(Task)

    if status is not None:
        stmt = stmt.where(Task.status == status)

    if cursor is not None:
        keys, direction = parse_dynamic_cursor(cursor, sort, order)
        cursor_sort_val, cursor_id = keys

        # Tuple comparison consistent with the sort's direction
        if order == SortOrder.desc:
            comparator = "<" if direction == "next" else ">"
        else:
            comparator = ">" if direction == "next" else "<"

        if comparator == "<":
            stmt = stmt.where(tuple_(sort_col, Task.id) < tuple_(cursor_sort_val, cursor_id))
        else:
            stmt = stmt.where(tuple_(sort_col, Task.id) > tuple_(cursor_sort_val, cursor_id))
    else:
        direction = "next"

    # Order
    if (order == SortOrder.desc) == (direction == "next"):
        # next + desc, or prev + asc → DESC order
        stmt = stmt.order_by(sort_col.desc(), Task.id.desc())
    else:
        stmt = stmt.order_by(sort_col.asc(), Task.id.asc())

    stmt = stmt.limit(limit + 1)
    result = await session.execute(stmt)
    rows = list(result.scalars().all())

    has_more = len(rows) > limit
    page_items = rows[:limit]

    # If we came in the opposite direction to the original sort, reverse the result
    if (order == SortOrder.desc) != (direction == "next"):
        page_items.reverse()

    # Generate the cursors...
    next_cursor = None
    previous_cursor = None
    if page_items:
        first, last = page_items[0], page_items[-1]
        if has_more or direction == "prev":
            next_cursor = make_dynamic_cursor(
                sort, order,
                (getattr(last, sort.value), last.id),
                "next",
            )
        if cursor is not None or direction == "prev":
            previous_cursor = make_dynamic_cursor(
                sort, order,
                (getattr(first, sort.value), first.id),
                "prev",
            )

    return Page[TaskOut](
        items=[TaskOut.model_validate(t) for t in page_items],
        next_cursor=next_cursor,
        previous_cursor=previous_cursor,
        has_more=has_more,
    )

Why does this matter in real work?

1. Real APIs have filters + sort + pagination simultaneously. "Just cursor pagination" works for tutorials. For production, you need to combine it with filters (by state, by owner, by date) and a dynamic sort. If your implementation doesn't handle them correctly, you'll be reimplementing pagination every 6 months.

2. SQL injection with identifiers is real. Plenty of people think "I use an ORM, I'm safe." But the ORM only parameterizes values, not column names. If you let the client specify columns via a string, you're vulnerable. An allowlist with an Enum is the correct defense — and FastAPI makes it almost automatic with type validation.

3. Index design. Every dynamic sort you offer requires a composite index. If your API allows sorting by 5 different columns, you need 5 composite indexes (each with the sort column + id). This isn't free: indexes take space and slow down writes. There's a balance — don't expose sorting on columns nobody will use.

4. Telling features apart from technical debt. "Allow sorting by any column" sounds flexible, but it generates debt. Every dynamic sort:

• Requires an index (storage + write cost).
• Requires testing the cursor (each sort is a separate case).
• Requires documentation.
• Requires the client to understand which sort to use.

Restricting the sort to 2-3 useful options is better than offering 10. Infinite flexibility is technical debt dressed up as a feature.

Traps and common mistakes

Mistake 1 (security): allowing sort as an unvalidated string

Symptom: ?sort=id; DROP TABLE tasks; -- breaks the API.

Why it happens: string concatenation in SQL, even through text() or an f-string.

How to tell: search your code for f"ORDER BY {sort}" or text(f"..."). If you find that with client input, you're vulnerable.

How to fix it: an allowlist with an Enum + explicit mapping to SQLAlchemy columns. The Enum blocks arbitrary values at the FastAPI layer; the mapping is the second defense.

Mistake 2 (conceptual): putting filters inside the cursor

Symptom: your cursor is eyJ...{ti:7,filters:{status:'active'},...} — it includes filters.

Why it's wrong: it couples the cursor to the filters. If the client changes a filter, the "old" cursor keeps carrying the old filters. Confusing behavior, inconsistent with serious APIs (Stripe, GitHub).

How to fix it: filters go as separate query params. The client is responsible for keeping them consistent between pages. Your API treats every request as independent.

Mistake 3 (practical): not validating that the cursor belongs to the current sort

Symptom: the client changes ?sort=priority to ?sort=updated_at but keeps the old cursor=.... Your API runs a query with a cursor of (priority_value, id) but an ORDER BY updated_at. Strange results (it can return duplicates, skip items, or fail silently).

Why it happens: you forgot to validate that cursor.s == request.sort.

How to fix it: include s (sort field) and o (order) in the cursor. Validate on decode. Reject with HTTP 400 if it doesn't match.

Mistake 4 (performance): not creating indexes for the available sorts

Symptom: sorting by created_at is fast (it has a composite index), but sorting by updated_at is 50x slower because it falls into a Sort without an index.

Why it happens: you added a dynamic sort to the endpoint but didn't create indexes for every option.

How to fix it: one composite index per combination (sort_column DESC, id DESC). If your sort accepts 5 columns, you need 5 indexes (plus the filters).

CREATE INDEX idx_tasks_updated_at_id_desc ON tasks (updated_at DESC, id DESC);
CREATE INDEX idx_tasks_priority_id_desc   ON tasks (priority DESC, id DESC);
CREATE INDEX idx_tasks_title_id_asc       ON tasks (title ASC, id ASC);

Mistake 5 (UX): changing the sort doesn't reset the cursor in the frontend

Symptom: the UI has a sort dropdown. The user changes the sort, the frontend makes a request with the new sort + the old cursor, the API returns 400.

Why it happens: the frontend kept the cursor in state when the sort changed.

How to fix it: a clear convention — when the sort, the filters, or any query param other than the cursor changes, the frontend discards the cursor. Document this in /docs.

Mistake 6 (conceptual): allowing sorting on columns that aren't unique and aren't tiebreakable

Symptom: ?sort=status (an enum with 3 values). The cursor carries (status_value, id). It technically works, but the client sees "all rows with status=open first, then all with status=in_progress, then closed." That's not what the UX wants.

Why it happens: sorting by low-cardinality columns groups all rows with the same value before moving to the next. It's valid SQL, bad UX.

How to fix it: restrict the dynamic sort to columns with enough cardinality for "sorting" to make sense (timestamps, names, IDs, scores). For categorical columns, add them as a filter, not as a sort.

Exercises

Exercise 1: add a created_after filter to the endpoint

Add an optional created_after query param (datetime) that filters tasks created after that date. Make sure it combines with the cursor correctly.

# app/main.py
from datetime import datetime

@app.get("/tasks")
async def get_tasks(
    cursor: str | None = Query(default=None),
    limit: int = Query(default=50, ge=1, le=200),
    status: TaskStatus | None = Query(default=None),
    priority: TaskPriority | None = Query(default=None),
    created_after: datetime | None = Query(
        default=None,
        description="Filter tasks created after this date (ISO 8601)",
    ),
    session: AsyncSession = Depends(get_session),
) -> Page[TaskOut]:
    return await list_tasks_paginated(
        session,
        cursor=cursor,
        limit=limit,
        status=status,
        priority=priority,
        created_after=created_after,
    )

# app/repositories/tasks.py
async def list_tasks_paginated(
    session: AsyncSession,
    cursor: str | None = None,
    limit: int = 50,
    status: TaskStatus | None = None,
    priority: TaskPriority | None = None,
    created_after: datetime | None = None,
) -> Page[TaskOut]:
    stmt = select(Task).order_by(Task.created_at.desc(), Task.id.desc())

    if status is not None:
        stmt = stmt.where(Task.status == status)
    if priority is not None:
        stmt = stmt.where(Task.priority == priority)
    if created_after is not None:
        stmt = stmt.where(Task.created_at > created_after)

    # ... the rest is identical ...

# Tasks created in the last 7 days
curl 'http://localhost:8000/tasks?created_after=2026-04-25T00:00:00Z&limit=10' | jq

# Combined with status
curl 'http://localhost:8000/tasks?status=open&created_after=2026-04-25T00:00:00Z' | jq

async def test_created_after_filter(client, session):
    now = datetime.now(timezone.utc)
    # Insert 2 old tasks and 3 new ones
    old = [
        Task(title=f"old_{i}", created_at=now - timedelta(days=10 + i))
        for i in range(2)
    ]
    new = [
        Task(title=f"new_{i}", created_at=now - timedelta(hours=i))
        for i in range(3)
    ]
    session.add_all(old + new)
    await session.commit()

    cutoff = (now - timedelta(days=5)).isoformat()
    r = await client.get(f"/tasks?created_after={cutoff}&limit=10")
    body = r.json()
    assert len(body["items"]) == 3
    assert all("new_" in t["title"] for t in body["items"])

Exercise 2: implement a dynamic sort with an allowlist

Add sort (enum: created_at, priority, title) and order (enum: asc, desc) to the endpoint. Make sure SQL injection is impossible.

# app/schemas.py
import enum

class TaskSortField(str, enum.Enum):
    created_at = "created_at"
    priority = "priority"
    title = "title"


class SortOrder(str, enum.Enum):
    asc = "asc"
    desc = "desc"

# app/repositories/tasks.py
from sqlalchemy import asc, desc

SORT_COLUMN_MAP = {
    TaskSortField.created_at: Task.created_at,
    TaskSortField.priority: Task.priority,
    TaskSortField.title: Task.title,
}


async def list_tasks_with_sort(
    session, sort, order, cursor, limit, status=None,
):
    sort_col = SORT_COLUMN_MAP[sort]  # KeyError if it isn't there → safe
    direction_fn = desc if order == SortOrder.desc else asc

    stmt = select(Task).order_by(direction_fn(sort_col), direction_fn(Task.id))

    if status is not None:
        stmt = stmt.where(Task.status == status)

    # Cursor with sort/order validation...
    if cursor is not None:
        keys, dir_cursor = parse_dynamic_cursor(cursor, sort, order)
        cursor_sort_val, cursor_id = keys
        if order == SortOrder.desc:
            stmt = stmt.where(
                tuple_(sort_col, Task.id) < tuple_(cursor_sort_val, cursor_id)
            )
        else:
            stmt = stmt.where(
                tuple_(sort_col, Task.id) > tuple_(cursor_sort_val, cursor_id)
            )

    stmt = stmt.limit(limit + 1)
    # ... the rest is the same

# Endpoint
@app.get("/tasks")
async def get_tasks(
    sort: TaskSortField = Query(default=TaskSortField.created_at),
    order: SortOrder = Query(default=SortOrder.desc),
    ...
):
    ...

async def test_sort_injection_is_impossible(client):
    # Attempt SQL injection
    r = await client.get("/tasks?sort=id;%20DROP%20TABLE%20tasks;%20--")
    assert r.status_code == 422
    # The error is Enum validation, it never reached the DB
    assert "Input should be" in str(r.json())

    # Try with a valid enum value
    r = await client.get("/tasks?sort=priority")
    assert r.status_code == 200

async def test_sort_by_priority(client, session):
    # Insert 3 tasks with different priorities
    session.add_all([
        Task(title="low_task", priority=TaskPriority.low),
        Task(title="high_task", priority=TaskPriority.high),
        Task(title="med_task", priority=TaskPriority.medium),
    ])
    await session.commit()

    r = await client.get("/tasks?sort=priority&order=desc&limit=5")
    body = r.json()
    priorities = [t["priority"] for t in body["items"]]
    assert priorities == ["low", "medium", "high"][::-1]  # DESC

Exercise 3: cursor validation by sort

Add validation that the cursor belongs to the request's current sort. Prove that changing the sort between pages returns HTTP 400.

# app/pagination_dynamic.py
def make_dynamic_cursor(sort, order, keys, direction):
    return encode_signed_cursor({
        "v": 1,
        "s": sort.value,
        "o": order.value,
        "k": list(keys),
        "d": direction,
    })


def parse_dynamic_cursor(cursor, expected_sort, expected_order):
    decoded = decode_signed_cursor(cursor)

    if decoded.get("s") != expected_sort.value:
        raise CursorError(
            f"Cursor belongs to sort='{decoded.get('s')}', "
            f"the request asks for sort='{expected_sort.value}'. "
            f"Restart pagination when changing the sort."
        )
    if decoded.get("o") != expected_order.value:
        raise CursorError(
            f"Cursor belongs to order='{decoded.get('o')}', "
            f"the request asks for order='{expected_order.value}'."
        )
    return decoded["k"], decoded.get("d", "next")

# Test
async def test_cursor_from_another_sort_is_rejected(client, session):
    # Seed
    session.add_all([Task(title=f"t_{i}") for i in range(5)])
    await session.commit()

    # Page 1 with sort=created_at
    r = await client.get("/tasks?sort=created_at&limit=2")
    cursor_created_at = r.json()["next_cursor"]

    # Try to use the cursor with sort=priority
    r = await client.get(
        f"/tasks?sort=priority&cursor={cursor_created_at}"
    )
    assert r.status_code == 400
    assert "sort='created_at'" in r.json()["detail"]
    assert "sort='priority'" in r.json()["detail"]


async def test_cursor_from_another_order_is_rejected(client, session):
    session.add_all([Task(title=f"t_{i}") for i in range(5)])
    await session.commit()

    r = await client.get("/tasks?sort=created_at&order=desc&limit=2")
    cursor = r.json()["next_cursor"]

    r = await client.get(
        f"/tasks?sort=created_at&order=asc&cursor={cursor}"
    )
    assert r.status_code == 400
    assert "order='desc'" in r.json()["detail"]

Exercise 4: identify missing indexes with EXPLAIN

You implemented sorting by created_at, priority, and title. Run EXPLAIN ANALYZE with each sort on a table with 100k rows. Identify which ones need a composite index and create them.

-- Test 1: sort by created_at (already has an index)
EXPLAIN (ANALYZE) SELECT * FROM tasks
ORDER BY created_at DESC, id DESC LIMIT 50;
-- Expected: Index Scan using idx_tasks_created_id_desc
-- Buffers: shared hit=4, Execution Time: <1ms ✅

-- Test 2: sort by priority (already has a composite index with a tiebreaker)
EXPLAIN (ANALYZE) SELECT * FROM tasks
ORDER BY priority DESC, id DESC LIMIT 50;
-- If you do NOT have idx_tasks_priority_id_desc:
-- → Sort + Seq Scan, Execution Time: 100+ms ❌
-- If you DO have it:
-- → Index Scan, Execution Time: <1ms ✅

-- Test 3: sort by title (probably has NO index)
EXPLAIN (ANALYZE) SELECT * FROM tasks
ORDER BY title ASC, id ASC LIMIT 50;
-- Expected without an index:
-- → Sort + Seq Scan, Execution Time: 200ms+ ❌

-- For sorting by priority
CREATE INDEX IF NOT EXISTS idx_tasks_priority_id_desc
  ON tasks (priority DESC, id DESC);

-- For sorting by priority ASC
-- (PostgreSQL can walk a DESC index in both directions,
--  so a single index serves both directions)

-- For sorting by title
CREATE INDEX IF NOT EXISTS idx_tasks_title_id_asc
  ON tasks (title ASC, id ASC);

ANALYZE tasks;

EXPLAIN (ANALYZE) SELECT * FROM tasks
ORDER BY title ASC, id ASC LIMIT 50;
-- Now: Index Scan using idx_tasks_title_id_asc, <1ms

Exercise 5: a design decision — which sorts to offer?

Your team is debating which sort options to expose in GET /tasks. The proposals:

a) "Only created_at (default DESC)." b) "created_at, priority, due_date." c) "created_at, priority, due_date, title, assignee_name, updated_at, status, id." d) "Any column — we pass a free-form sort=column_name."

Which do you choose and why? Consider: indexes, maintenance, UX.

Summary and next step

In this capsule you learned:

• Filters go OUTSIDE the cursor. The cursor only carries position. The client repeats the filters between pages.
• A dynamic sort is safe with an allowlist + Enum. Never concatenate strings in SQL — use an explicit mapping from enum to columns.
• The cursor carries the sort/order to validate consistency. If the client changes the sort, you reject with HTTP 400 — better than returning strange data.
• Every dynamic sort requires a composite index. Without one, that sort is 100x slower. Restricting to 2-3 useful options is a feature, not a limitation.
• FastAPI + a Pydantic Enum + SQLAlchemy's tuple_() is defense in depth: three layers an attacker has to break to inject SQL.

Before moving on you should be able to:

• Design an endpoint with 2-3 filters + 2-3 sort options without opening SQL injection vulnerabilities
• Justify why each sort needs its composite index
• Handle the "the client changed the sort between pages" case with a clear HTTP 400 error
• Decide when to restrict sorts vs when to add them

Next capsule — The module project: cursor pagination in TaskFlow. You're going to build a complete endpoint with everything you learned (an opaque HMAC cursor, bidirectional, filters, dynamic sort) on a table of 5M tasks. You'll measure benchmarks (page 1 vs page 50,000) and verify that the speedup vs OFFSET is real (~17x according to Design Gurus). The project consolidates the whole module and produces the BENCHMARKS.md that proves you understand cursor pagination end to end.

Resources

1. OWASP — SQL Injection Prevention Cheat Sheet — the canonical reference. Note the "Defense Option 4: Escaping All User Supplied Input" section, which explains why identifiers can't be parameterized.
2. SQLAlchemy 2.0 — Dynamic Sorting — the official reference for programmatic asc() and desc().
3. FastAPI — Query Parameters with Enums — how to validate query params with an Enum.
4. Stripe API — Filtering and Pagination — a case study: how Stripe combines filters (customer, created, status) with a cursor (starting_after, ending_before).
5. Brandur Leach — "API Versioning" — discusses the trade-offs of exposing the internal schema (sorting by columns) vs hiding it.
6. Markus Winand — "Two-Phase Cursor Pagination" — includes a discussion of how to combine filters with keyset pagination.
7. PostgreSQL Documentation — Multicolumn Indexes — the official reference on when to use composite indexes.
8. Milan Jovanović — "Filtering, Sorting and Pagination in EF Core" — the equivalent pattern in .NET, transferable concepts.

Module 1 — SQL Patterns for Production APIs Guide

Next capsule: The module project — cursor pagination in TaskFlow with measured benchmarks.