Module 1: Pagination Patterns
Cursor pagination: fundamentals
Capsule overview
In the previous capsule you saw why OFFSET is O(n) and breaks on deep pages. This capsule gives you the replacement: cursor pagination. The core idea is simple, but the details matter — and most tutorials skip right over them.
You're going to learn what a cursor is exactly (it isn't the same as keyset, even though the two get mixed up), how it's encoded so the client treats it as opaque, what minimum information it needs to guarantee stable pagination, and how to model the response with Pydantic v2 so your API feels consistent with serious APIs (Stripe, GitHub, Slack).
This capsule is conceptual and introduces the model. The end-to-end implementation with SQLAlchemy 2.0 async + FastAPI comes in the next capsule (04). Here you build the mental model — no endpoint code yet, but you do get encoding/decoding code and Pydantic models you'll reuse.
The concept: navigate by value, not by position
OFFSET asks PostgreSQL: "skip the first 50,000 rows for me." Cursor asks: "give me the first 50 rows that come after this exact point."
OFFSET (positional):
"Give me 50 rows, skipping the first 50,000"
→ PostgreSQL reads 50,050 rows from the index. O(n).
Cursor (value-based):
"Give me 50 rows where created_at < '2026-04-15 10:23:45+00'"
→ PostgreSQL jumps straight to that position in the index. O(log n).
The difference is radical. A B-tree index can do a binary lookup to find the position of '2026-04-15 10:23:45+00' in O(log n). Once there, it reads the next 50 rows — no matter how "deep" you are in the dataset, the cost is constant.
Mental model: the book bookmark
Imagine a 5,000-page book. You want to read 50 pages starting from page 4,500.
OFFSET is like counting pages with your finger: you open the book, count 1, 2, 3 ... 4,499, and start reading. It takes a while.
Cursor is like a bookmark: someone tells you "open the book where the paragraph's date is April 15." You open the book almost directly at that spot (books have alphabetical indexes by date at the back). You start reading. It's fast no matter how deep you are.
┌──────────────────────────────────────────────────────────────┐
│ Table `tasks` ordered by created_at DESC │
│ with index idx_tasks_created_at_desc │
│ │
│ Row 1 created_at = 2026-05-01 18:30:00 │
│ Row 2 created_at = 2026-05-01 18:29:55 │
│ ... │
│ Row N created_at = 2026-04-15 10:23:45 ← cursor points here│
│ Row N+1 created_at = 2026-04-15 10:23:40 │
│ Row N+2 created_at = 2026-04-15 10:23:35 │
│ ... │
│ │
│ Query with cursor: │
│ WHERE created_at < '2026-04-15 10:23:45' │
│ ORDER BY created_at DESC LIMIT 50 │
│ │
│ → PostgreSQL: binary lookup in the B-tree (O(log n)), │
│ then reads 50 consecutive rows. Total: O(log n + page_size)│
└──────────────────────────────────────────────────────────────┘
That's the one and only reason cursor pagination is 17x faster on page 50,000: it turns O(n) into O(log n + 50). The "+50" is constant (the LIMIT), and log n is ~22 for a 5M-row table. Latency is almost insensitive to the depth of the page.
Cursor vs keyset: the difference almost nobody clarifies
These two words get used as synonyms on the internet, but they are not the same thing. The distinction matters.
Keyset pagination
Keyset is the SQL technique: the client hands back the real values of the sort columns, and the API builds a WHERE with tuple comparison.
Client: "give me the next 50 tasks created before 2026-04-15 10:23:45,
with id lower than 12345 if there's a timestamp tie"
API: SELECT ... WHERE (created_at, id) < ('2026-04-15 10:23:45', 12345)
ORDER BY created_at DESC, id DESC LIMIT 50
Pros:
- Readable URL:
?after_created_at=2026-04-15T10:23:45Z&after_id=12345 - The client can build the cursor on its own (it knows which columns you use).
Cons:
- It leaks schema details to the client. If tomorrow you want to change the sort from
(created_at, id)to(updated_at, id), you break your consumers. - The client can manipulate the values and read data it shouldn't (e.g. put
tenant_idin the cursor and change it). - Ugly, long URLs with lots of parameters.
Cursor pagination
Cursor is the API technique: the client hands back an opaque token (an encoded string) that the API decodes internally to run the query. The client neither knows nor cares what's inside the cursor.
Client: "give me the next 50 tasks after the cursor 'eyJ0Ijo...'"
API:
1. Decodes the cursor: {"created_at": "2026-04-15T10:23:45Z", "id": 12345}
2. Builds the keyset query internally:
SELECT ... WHERE (created_at, id) < ('2026-04-15...', 12345)
ORDER BY created_at DESC, id DESC LIMIT 50
3. Generates the next cursor from the last returned row.
Pros:
- The client knows nothing about the internal schema. You can change the sort from
(created_at, id)to(updated_at, id)without breaking consumers (as long as the old cursor still decodes or is explicitly invalidated). - You can sign the cursor with HMAC to prevent tampering (capsule 06).
- Clean URLs:
?cursor=eyJ0Ijo....
Cons:
- Requires encoding/decoding (10-20 lines of code).
- The client can't "build" a cursor by hand — it only receives the next one from the response.
The relationship between the two
Cursor pagination is almost always implemented with keyset pagination internally. The cursor is just the opaque wrapper around the keyset.
┌─────────────────────────────────────────────┐
│ Client receives: cursor = "eyJ0Ijo..." │
│ Client sends back: ?cursor=eyJ0Ijo... │
└──────────────────┬──────────────────────────┘
│
▼
┌─────────────────────────────────────────────┐
│ API decodes the cursor: │
│ {"created_at": "...", "id": 12345} │
└──────────────────┬──────────────────────────┘
│
▼
┌─────────────────────────────────────────────┐
│ API runs the keyset query: │
│ WHERE (created_at, id) < (..., 12345) │
│ ORDER BY created_at DESC, id DESC │
│ LIMIT 50 │
└─────────────────────────────────────────────┘
This module's recommendation: use cursor (an opaque token) for public APIs or when there are external clients. Use pure keyset when it's an internal API between services where you control both sides and simplicity wins.
Anatomy of a cursor
A cursor encodes all the information needed to make the next query stable. What is that?
For a sort on (created_at DESC, id DESC), you need:
- The
created_atof the last returned row. - The
idof the last returned row (the tiebreaker).
That's all for the simple case. For more complex cases:
- Dynamic sort (e.g. the user can pick between
created_at,priority,updated_at): include which column was used. - Filters (e.g.
?status=active): do NOT include filters in the cursor — they're part of the query. - Direction (next vs previous): include a direction flag (capsule 06).
- Expiration: optional, you can include a cursor-creation timestamp.
Recommended structure
{
"v": 1, # cursor schema version (for evolution)
"t": "2026-04-15T10:23:45.123Z", # timestamp of the last row (created_at)
"i": 12345, # id of the last row (tiebreaker)
"d": "next" # direction: "next" or "prev"
}
The version matters: if tomorrow you change the cursor format, you can reject old versions with a clear error instead of returning incorrect data.
Encoding: URL-safe base64
So the cursor can be passed in a URL without trouble, encode it with URL-safe base64:
import base64
import json
def encode_cursor(payload: dict) -> str:
"""Encodes a dict as a URL-safe opaque cursor."""
raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8")
# urlsafe_b64encode uses - and _ instead of + and /, and needs no URL escaping
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
def decode_cursor(cursor: str) -> dict:
"""Decodes an opaque cursor back into a dict."""
# Restore the padding (urlsafe_b64encode strips it)
padding = "=" * (-len(cursor) % 4)
raw = base64.urlsafe_b64decode(cursor + padding)
return json.loads(raw)
Example usage:
payload = {
"v": 1,
"t": "2026-04-15T10:23:45.123Z",
"i": 12345,
"d": "next"
}
cursor = encode_cursor(payload)
print(cursor)
# eyJkIjoibmV4dCIsImkiOjEyMzQ1LCJ0IjoiMjAyNi0wNC0xNVQxMDoyMzo0NS4xMjNaIiwidiI6MX0
decoded = decode_cursor(cursor)
print(decoded)
# {'d': 'next', 'i': 12345, 't': '2026-04-15T10:23:45.123Z', 'v': 1}
Important notes:
- This isn't encryption. Anyone can decode this cursor with base64. If you need security (so the client can't tamper with the cursor), add HMAC — we cover it in capsule 06.
separators=(",", ":")removes unnecessary whitespace and keeps the cursor compact (~50-80 characters).sort_keys=Trueguarantees the cursor is deterministic (always the same cursor for the same payload). Useful for tests.
Modeling the paginated response with Pydantic v2
A well-designed paginated response includes:
- The list of items.
- The cursor for the next page.
- A
has_moreflag so the client knows whether there's more without having to ask. - (Optional) A cursor for the previous page — bidirectional, capsule 06.
Generic Page[T] model
# pagination/models.py
from typing import Generic, TypeVar
from pydantic import BaseModel, Field
T = TypeVar("T")
class Page(BaseModel, Generic[T]):
"""Generic paginated response using 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"
)
model_config = {"arbitrary_types_allowed": True}
Usage with a concrete model
# tasks/schemas.py
from datetime import datetime
from pydantic import BaseModel
from pagination.models import Page
class TaskOut(BaseModel):
id: int
title: str
created_at: datetime
model_config = {"from_attributes": True} # so the SQLAlchemy ORM works
# The endpoint's response
TaskPage = Page[TaskOut]
Example JSON response
{
"items": [
{"id": 12345, "title": "Review PR #42", "created_at": "2026-04-15T10:23:45.123Z"},
{"id": 12344, "title": "Update Slack channel", "created_at": "2026-04-15T10:22:30.000Z"},
{"id": 12343, "title": "Sync with design", "created_at": "2026-04-15T10:20:12.500Z"}
],
"next_cursor": "eyJ0IjoiMjAyNi0wNC0xNVQxMDoyMDoxMi41MDBaIiwiaSI6MTIzNDMsImQiOiJuZXh0IiwidiI6MX0",
"has_more": true
}
The client uses next_cursor in the next request: GET /tasks?cursor=eyJ0IjoiMjAyNi0w....
How to determine has_more correctly
There are two approaches:
Approach 1: ask for LIMIT + 1 rows.
LIMIT_PLUS_ONE = 51 # if page_size = 50
# You ask for 51 rows. If you get 51, there's more; you return only the first 50.
# If you get ≤50, there's no more.
Pros: a single query. Cons: you have to discard the extra row before returning to the client.
Approach 2: run a separate query to find out whether there's more.
# You ask for 50 rows. Then you run a secondary query:
# SELECT 1 FROM tasks WHERE (created_at, id) < (cursor.t, cursor.i) LIMIT 1
Pros: clean separation. Cons: two queries (an extra round-trip).
This guide's recommendation: approach 1 (LIMIT + 1). It's the de facto standard and it avoids the extra round-trip. The complexity of discarding the extra row is trivial.
Worked example: a complete encoding round-trip
We're going to do the full cycle: encode a cursor → decode → build the SQL query → return the response.
Minimal setup
# round_trip_demo.py
"""
Demo of the cursor cycle: encode → decode → query → response.
No SQLAlchemy yet, just the conceptual model.
"""
import base64
import json
from datetime import datetime, timezone
from pydantic import BaseModel
# --- 1. Encoding helpers ---
def encode_cursor(payload: dict) -> str:
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:
padding = "=" * (-len(cursor) % 4)
raw = base64.urlsafe_b64decode(cursor + padding)
return json.loads(raw)
# --- 2. Response model ---
class TaskOut(BaseModel):
id: int
title: str
created_at: datetime
class TaskPage(BaseModel):
items: list[TaskOut]
next_cursor: str | None
has_more: bool
# --- 3. Simulating "data in the DB" ---
# We simulate 5 tasks ordered by created_at DESC
fake_db = [
TaskOut(id=12345, title="Task A", created_at=datetime(2026, 4, 15, 10, 23, 45, tzinfo=timezone.utc)),
TaskOut(id=12344, title="Task B", created_at=datetime(2026, 4, 15, 10, 22, 30, tzinfo=timezone.utc)),
TaskOut(id=12343, title="Task C", created_at=datetime(2026, 4, 15, 10, 20, 12, tzinfo=timezone.utc)),
TaskOut(id=12342, title="Task D", created_at=datetime(2026, 4, 15, 10, 18, 5, tzinfo=timezone.utc)),
TaskOut(id=12341, title="Task E", created_at=datetime(2026, 4, 15, 10, 15, 0, tzinfo=timezone.utc)),
]
# --- 4. Pagination function ---
def paginate_tasks(cursor: str | None, page_size: int = 2) -> TaskPage:
"""Paginates tasks using an opaque cursor. Page size = 2 for the demo."""
# 4a. Decode the cursor (if one came in) or start from the beginning
if cursor:
decoded = decode_cursor(cursor)
last_t = datetime.fromisoformat(decoded["t"])
last_id = decoded["i"]
# Filter items that come AFTER the cursor (tuple comparison)
candidates = [
t for t in fake_db
if (t.created_at, t.id) < (last_t, last_id)
]
else:
candidates = fake_db
# 4b. Take page_size + 1 to find out whether there's more
fetched = candidates[: page_size + 1]
has_more = len(fetched) > page_size
items = fetched[:page_size] # discard the extra row
# 4c. Generate the cursor from the last returned item
next_cursor = None
if has_more and items:
last_item = items[-1]
payload = {
"v": 1,
"t": last_item.created_at.isoformat().replace("+00:00", "Z"),
"i": last_item.id,
"d": "next"
}
next_cursor = encode_cursor(payload)
return TaskPage(items=items, next_cursor=next_cursor, has_more=has_more)
# --- 5. Demo of the full flow ---
if __name__ == "__main__":
print("=== Page 1 (no cursor) ===")
page1 = paginate_tasks(cursor=None, page_size=2)
print(f"Items: {[t.title for t in page1.items]}")
print(f"next_cursor: {page1.next_cursor}")
print(f"has_more: {page1.has_more}\n")
print("=== Page 2 (with page 1's cursor) ===")
page2 = paginate_tasks(cursor=page1.next_cursor, page_size=2)
print(f"Items: {[t.title for t in page2.items]}")
print(f"next_cursor: {page2.next_cursor}")
print(f"has_more: {page2.has_more}\n")
print("=== Page 3 (with page 2's cursor) ===")
page3 = paginate_tasks(cursor=page2.next_cursor, page_size=2)
print(f"Items: {[t.title for t in page3.items]}")
print(f"next_cursor: {page3.next_cursor}")
print(f"has_more: {page3.has_more}")
To run it:
pip install pydantic
python round_trip_demo.py
Expected output:
=== Page 1 (no cursor) ===
Items: ['Task A', 'Task B']
next_cursor: eyJkIjoibmV4dCIsImkiOjEyMzQ0LCJ0IjoiMjAyNi0wNC0xNVQxMDoyMjozMFoiLCJ2IjoxfQ
has_more: True
=== Page 2 (with page 1's cursor) ===
Items: ['Task C', 'Task D']
next_cursor: eyJkIjoibmV4dCIsImkiOjEyMzQyLCJ0IjoiMjAyNi0wNC0xNVQxMDoxODowNVoiLCJ2IjoxfQ
has_more: True
=== Page 3 (with page 2's cursor) ===
Items: ['Task E']
next_cursor: None
has_more: False
Reading it:
- Page 1 returned Task A and B. There's more (
has_more=True), and the cursor points to the last returned item (Task B with id=12344). - Page 2 decoded the cursor, filtered items with
(created_at, id) < (cursor.t, cursor.i), and returned Task C and D. - Page 3 returned only Task E (the only one left).
has_more=Falseandnext_cursor=None.
That's the complete conceptual cycle. In capsule 04 you're going to implement the same thing but with SQLAlchemy 2.0 async + FastAPI + real PostgreSQL. The logic is identical — the only thing that changes is "candidates = an in-memory list" becoming a SQL query.
Why does this matter in real work?
1. Public APIs with a latency SLA.
Stripe guarantees its API responds in <200ms p99. How do they pull that off with customers who have millions of transactions? Cursor pagination on (created_at, id). It's an architectural decision that scales. If you're going to design a public API over any dataset that grows, cursor is the default choice.
2. Backfills and sync APIs. Any "export every record since date X" integration uses a cursor implicitly or explicitly. The client wants to "read the whole history without the session breaking." Cursor is stable under insertions — new records don't wedge themselves into the flow.
3. Mobile apps with infinite scroll. Twitter, Instagram, Slack — every modern feed uses cursor. The reason isn't just performance — it's that cursor is stable under insertions. When someone posts a new tweet while you're scrolling, cursor guarantees you don't see the same post twice. OFFSET would.
4. Senior vs junior differentiation.
"I implemented cursor pagination with base64 encoding and composite cursors on (created_at, id) to guarantee pagination that's stable under insertions" is a sentence that separates a senior from a junior in any SaaS interview. Most people know LIMIT/OFFSET. Few can articulate cursor properly.
Traps and common mistakes
Mistake 1 (conceptual): confusing cursor with keyset
Symptom: "I call the URL ?after_created_at=2026-04-15T10:23:45Z&after_id=12345 a cursor."
Why it's confusing: internally the query is the same (keyset). But at the API level they're different things: a cursor is an opaque token; keyset is direct exposure of the values. The difference matters when security or schema evolution is on the table.
How to tell: if the client can read the cursor and understand what each part means, it isn't a cursor — it's keyset. Cursor implies opacity.
Mistake 2 (practical): returning the cursor in the response as a JSON object instead of a string
Symptom: your response returns "next_cursor": {"t": "2026-04-15T...", "i": 12345}.
Why it's wrong: it breaks the "the client doesn't need to understand what's inside" abstraction. You're leaking the internal schema. If tomorrow you change the cursor's field names, you break every consumer.
How to fix it: always return the cursor as an opaque string (base64). The client treats it as a black box: receives it, stores it, hands it back. Period.
Mistake 3 (conceptual): not including a version in the cursor
Symptom: six months later you need to change the cursor format (add a field). Old cursors saved in customers' URLs start breaking silently.
Why it happens: without a version, you can't know whether a received cursor is in the current format or the old one. The only option is to decode and pray.
How to fix it: include "v": 1 in every cursor. When you change the format, bump it to "v": 2. The API rejects cursors with an unknown v, or keeps supporting the old ones for a grace period.
def decode_cursor(cursor: str) -> dict:
decoded = ... # base64 decode + json.loads
if decoded.get("v") != 1:
raise ValueError(f"Unsupported cursor version: {decoded.get('v')}")
return decoded
Mistake 4 (edge case): a cursor without a tiebreaker on a non-unique sort
Symptom: your cursor is {"t": "2026-04-15T10:23:45Z"} (timestamp only). There are three tasks with that exact same timestamp. Pagination returns them in an unstable order and they get duplicated.
Why it happens: without a tiebreaker (id), the WHERE created_at < $1 returns every row with a lower timestamp — but rows with the same timestamp as the cursor can land inside or outside depending on the implementation.
How to fix it: always include the PK as a tiebreaker. Minimum cursor: {"t": "...", "i": 12345}. Query: WHERE (created_at, id) < ($1, $2). This is covered in detail in capsule 05.
Mistake 5 (practical): returning has_more=true without generating a next_cursor
Symptom: your response has has_more: true but next_cursor: null. The client doesn't know how to ask for the next page.
Why it happens: a bug in the wrapper's logic. The universal rule: has_more=true ⟺ next_cursor != null. They're the same information expressed two ways.
How to fix it: an explicit test in CI:
def test_has_more_iff_next_cursor():
page = paginate_tasks(cursor=None, page_size=2)
assert page.has_more == (page.next_cursor is not None)
Exercises
Exercise 1: encode/decode round-trip
Implement the encode_cursor and decode_cursor functions (the ones from the worked example). Prove that decode(encode(payload)) == payload for various payloads.
See solution
# test_cursor_roundtrip.py
import base64
import json
import pytest
def encode_cursor(payload: dict) -> str:
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:
padding = "=" * (-len(cursor) % 4)
raw = base64.urlsafe_b64decode(cursor + padding)
return json.loads(raw)
@pytest.mark.parametrize("payload", [
{"v": 1, "t": "2026-04-15T10:23:45Z", "i": 12345, "d": "next"},
{"v": 1, "t": "2026-01-01T00:00:00Z", "i": 1, "d": "next"},
{"v": 1, "t": "2026-12-31T23:59:59.999Z", "i": 999999999, "d": "prev"},
{"v": 2, "t": "2026-06-15T12:00:00Z", "i": 0, "d": "next", "extra": "value"},
])
def test_roundtrip(payload):
encoded = encode_cursor(payload)
decoded = decode_cursor(encoded)
assert decoded == payload
# The cursor must be URL-safe (no + / = characters)
assert "+" not in encoded
assert "/" not in encoded
assert "=" not in encoded
def test_cursor_is_compact():
"""Verifies the cursor is reasonably compact."""
payload = {"v": 1, "t": "2026-04-15T10:23:45.123Z", "i": 12345, "d": "next"}
cursor = encode_cursor(payload)
assert len(cursor) < 100 # typically ~80 chars
pip install pytest
pytest test_cursor_roundtrip.py -v
# 5 passed in 0.04s
Why it works: urlsafe_b64encode uses - and _ instead of + and /, and we strip the = padding so the cursor is clean in URLs. Decoding restores the padding to the nearest multiple of 4.
Exercise 2: reject cursors with an incompatible version
Modify decode_cursor so it raises ValueError if the cursor's version isn't 1. Test with a v=2 cursor that must be rejected.
See solution
# cursor_versioned.py
import base64
import json
CURRENT_VERSION = 1
def decode_cursor(cursor: str) -> dict:
padding = "=" * (-len(cursor) % 4)
try:
raw = base64.urlsafe_b64decode(cursor + padding)
decoded = json.loads(raw)
except (ValueError, json.JSONDecodeError) as e:
raise ValueError(f"Invalid cursor: malformed") from e
version = decoded.get("v")
if version != CURRENT_VERSION:
raise ValueError(
f"Cursor version {version} not supported (current: v{CURRENT_VERSION})"
)
return decoded
# Test
import pytest
def test_rejects_unknown_version():
# Build a v=2 cursor by hand
bad_payload = {"v": 2, "t": "2026-04-15T10:23:45Z", "i": 12345}
raw = json.dumps(bad_payload).encode("utf-8")
bad_cursor = base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
with pytest.raises(ValueError, match="version 2"):
decode_cursor(bad_cursor)
def test_rejects_malformed_cursor():
with pytest.raises(ValueError, match="malformed"):
decode_cursor("not_valid_base64_!!!")
Why it works: validating the version explicitly at decode time gives you a single place to manage the cursor's evolution. When you add v=2, you can decide: reject v=1 (forcing re-pagination), support them temporarily, or migrate them. Without a version, that decision is impossible to make.
Recommended production pattern: log every time a cursor with an old version arrives, so you know when it's safe to remove it.
Exercise 3: implement paginate_in_memory with tuple comparison
Take the list of 5 tasks from the worked example. Implement paginate_in_memory(items, cursor, page_size) that paginates using tuple comparison. Test all three pages.
See solution
# paginate_in_memory.py
from datetime import datetime, timezone
from typing import TypeVar, Callable
import base64, json
T = TypeVar("T")
def encode_cursor(payload: dict) -> str:
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:
padding = "=" * (-len(cursor) % 4)
raw = base64.urlsafe_b64decode(cursor + padding)
return json.loads(raw)
def paginate_in_memory(
items: list[T],
cursor: str | None,
page_size: int,
sort_key: Callable[[T], tuple],
) -> dict:
"""
Paginates an in-memory list using a cursor.
Args:
items: list pre-sorted by sort_key descending.
cursor: opaque cursor (None for the first page).
page_size: number of items per page.
sort_key: function that extracts (timestamp, id) from the item.
"""
if cursor:
decoded = decode_cursor(cursor)
cursor_tuple = (datetime.fromisoformat(decoded["t"]), decoded["i"])
# Filter items that come after (tuple comparison)
candidates = [it for it in items if sort_key(it) < cursor_tuple]
else:
candidates = items
fetched = candidates[: page_size + 1]
has_more = len(fetched) > page_size
page_items = fetched[:page_size]
next_cursor = None
if has_more and page_items:
last = page_items[-1]
last_t, last_i = sort_key(last)
next_cursor = encode_cursor({
"v": 1,
"t": last_t.isoformat().replace("+00:00", "Z"),
"i": last_i,
"d": "next"
})
return {
"items": page_items,
"next_cursor": next_cursor,
"has_more": has_more,
}
# Test
class Task:
def __init__(self, id, title, created_at):
self.id = id
self.title = title
self.created_at = created_at
fake_db = [
Task(12345, "A", datetime(2026, 4, 15, 10, 23, 45, tzinfo=timezone.utc)),
Task(12344, "B", datetime(2026, 4, 15, 10, 22, 30, tzinfo=timezone.utc)),
Task(12343, "C", datetime(2026, 4, 15, 10, 20, 12, tzinfo=timezone.utc)),
Task(12342, "D", datetime(2026, 4, 15, 10, 18, 5, tzinfo=timezone.utc)),
Task(12341, "E", datetime(2026, 4, 15, 10, 15, 0, tzinfo=timezone.utc)),
]
def test_three_pages():
sk = lambda t: (t.created_at, t.id)
p1 = paginate_in_memory(fake_db, None, 2, sk)
assert [t.title for t in p1["items"]] == ["A", "B"]
assert p1["has_more"] is True
assert p1["next_cursor"] is not None
p2 = paginate_in_memory(fake_db, p1["next_cursor"], 2, sk)
assert [t.title for t in p2["items"]] == ["C", "D"]
assert p2["has_more"] is True
p3 = paginate_in_memory(fake_db, p2["next_cursor"], 2, sk)
assert [t.title for t in p3["items"]] == ["E"]
assert p3["has_more"] is False
assert p3["next_cursor"] is None
Why it works: (t.created_at, t.id) < (cursor_t, cursor_i) is Python tuple comparison — it works element by element, just like in SQL. For different created_at values, it compares timestamps. For equal created_at values, it uses id as the tiebreaker. It's exactly what you're going to do in SQL in capsule 05.
Exercise 4: prove that cursor is stable under insertions
Take the 5 tasks from the example. Ask for page 1 with paginate_in_memory (it returns A, B and a cursor). Then insert a new task at the top:
new_task = Task(99999, "Z", datetime(2026, 5, 1, 12, 0, 0, tzinfo=timezone.utc))
fake_db.insert(0, new_task)
Ask for page 2 with page 1's cursor. Does the new task Z show up or not? Why?
See solution
def test_stable_under_insertions():
sk = lambda t: (t.created_at, t.id)
# Page 1: A, B
p1 = paginate_in_memory(fake_db, None, 2, sk)
assert [t.title for t in p1["items"]] == ["A", "B"]
# Insert a new task at the top (most recent)
new_task = Task(99999, "Z", datetime(2026, 5, 1, 12, 0, 0, tzinfo=timezone.utc))
fake_db.insert(0, new_task)
# Page 2 with page 1's cursor
p2 = paginate_in_memory(fake_db, p1["next_cursor"], 2, sk)
assert [t.title for t in p2["items"]] == ["C", "D"] # does NOT include Z
# And why? Because the cursor points at "B" (id=12344, t=10:22:30).
# The query is WHERE (created_at, id) < (10:22:30, 12344).
# Z has created_at=12:00:00 (more recent than B), so it does NOT match the WHERE.
# Z stays out of page 2 — the client will see it if they refresh page 1.
Analysis:
- Cursor pagination is stable under recent insertions: new items don't wedge themselves into pages the client is already scrolling.
- This is the reason Twitter, Instagram, and Slack use cursor: when someone posts while you're scrolling, you don't see duplicates or the same item twice.
- Compared with OFFSET: if this were OFFSET 2 LIMIT 2, page 2 would return [B, C] — that is, B repeated (because inserting Z pushed B to position 2). Cursor avoids that.
Trade-off: cursor is stable under insertions, but it isn't stable under deletions. If someone deletes B while you're on page 1 (which already returned B and C), your page 2 can start at D — skipping nothing, because you already saw B and C. But if someone deletes C between page 1 and page 2, page 2 can start at D and you lost C. This is acceptable in most cases (the "deletion" is generally real — the item is gone), but it has to be documented.
Summary and next step
In this capsule you learned:
- Cursor pagination navigates by value, not by position. It turns O(n) into O(log n + page_size).
- Cursor ≠ keyset. Cursor is the opaque wrapper; keyset is the internal SQL technique. Cursor = keyset + opacity + the ability to sign it.
- The minimum anatomy of a cursor: version + timestamp + id (tiebreaker) + direction.
- URL-safe base64 encoding keeps the cursor opaque to the client and URL-safe without escaping.
- Pydantic's
Page[T]gives you a consistent response:items,next_cursor,has_more. The de facto standard in serious APIs. has_moreis determined with LIMIT + 1 — you ask for one extra and discard it. A single query.- Cursor is stable under insertions: new items don't wedge themselves between pages the client is looking at.
Before moving on you should be able to:
- Explain the difference between cursor and keyset to a colleague
- Encode and decode a base64 cursor from memory
- Design the cursor payload structure for your case (which fields to include)
- Model
Page[T]with Pydantic v2
Next capsule — Cursor pagination in FastAPI. You're going to implement it end-to-end: SQLAlchemy 2.0 async with the keyset query, a FastAPI endpoint that accepts ?cursor=...&limit=..., input validation, error handling (invalid cursor → HTTP 400), and tests with httpx. Capsule 04 is pure code — you already have the conceptual logic here.
Resources
- Stripe API Reference — Pagination — the de facto standard. Stripe uses
starting_after/ending_beforewith an opaque cursor. A model of what cursor looks like in a serious public API. - Slack Engineering — "Evolving API Pagination at Slack" — a real case of migrating from OFFSET to cursor. Covers the problems of cursor evolution.
- GitHub REST API — Pagination — uses cursor with
Linkheaders (a REST variant). Worth it as a reference for a different approach. - Markus Winand — "Keyset Pagination" — the technical reference on keyset. Required reading to understand the SQL foundation.
- Brandur Leach — "API Paginations Design" — a discussion of cursor vs offset from Stripe's perspective (ex-employee).
- Pydantic v2 — Generics — the official documentation on Generics in Pydantic, which is what we use for
Page[T]. - Python
base64module — the official reference. The key functions areurlsafe_b64encode/urlsafe_b64decode. - Cursor Pagination spec — Relay GraphQL — the formal spec of cursor pagination in GraphQL. Even though we don't use GraphQL in the module, the spec is a good conceptual reference.
Module 1 — SQL Patterns for Production APIs Guide
Next capsule: Cursor pagination in FastAPI — end-to-end implementation with SQLAlchemy 2.0 async.