How to Replace OFFSET Pagination in a FastAPI API With Signed Keyset Cursors

What you will build, and why OFFSET paging breaks

Almost every API that lists things has to hand them out in pieces. A client asks for the first 50 events, then the next 50, and keeps going until it has seen them all. Each piece is called a page, and the technique is called pagination. The usual way to ask for the next page is to say how many rows to skip: LIMIT 50 OFFSET 100. It is easy to write and nearly every SQL database has a form of it. It also hides three problems that stay invisible on a small, quiet table and show up on a big, busy one.

The first problem is that pages drift. The client counts rows while the table keeps changing, so a new row can push an old one onto the next page, and a deletion can pull a row back onto a page the client has already read. Markus Winand, who created the SQL tuning site Use The Index, Luke, says it plainly in a post on keyset pagination: “The idea to use the number of rows seen to skip over them later is simply wrong.” The second problem is cost. The database has no shortcut to row 800,000; it has to produce those rows and throw them away, and the PostgreSQL manual says so in one sentence: “The rows skipped by an OFFSET clause still have to be computed inside the server; therefore a large OFFSET might be inefficient.” The third problem is exposure. If the client controls the offset and the page size, anyone can ask for the most expensive page there is, or for a million rows at once. OWASP’s page on unrestricted resource consumption lists “Number of records per page to return in a single request-response” among the limits an API has to define.

In this tutorial you will build a FastAPI endpoint, GET /events, that pages through a SQLite table of one million rows using keyset pagination: instead of “skip 800,000 rows”, the client says “give me the rows that come after the last one I saw.” You will reproduce each OFFSET failure with real numbers, watch a first keyset attempt lose 3.5 percent of the table, fix it, read the query plan to prove the database is not scanning, wrap the position in a signed cursor so clients cannot forge one without your key, and finish with a test suite. Every output below came from running the code on Windows 11 with Python 3.13.14 and SQLite 3.50.4. Your timings will differ; the shapes will not.

Nine terms to know before we start

Page and limit. A page is one batch of rows. The limit is the maximum number of rows the client wants in it.

Offset pagination. Asking for page N by telling the database to skip the first (N minus 1) times the page size rows.

Keyset pagination. Also called the seek method. The server remembers the sort key of the last row it returned and the next query asks only for rows that sort after it. Winand’s one-line recipe is to “just use a where clause that selects only data you haven’t seen yet.”

Cursor. The value the server hands back with each page so the client can ask for the next one. In this tutorial it is an opaque string called next_cursor. Opaque means the client is not supposed to look inside it.

Sort key and tiebreaker. The sort key is the column or columns that define the order, here the time an event happened. A tie is two rows with the same sort key value. SQLite’s documentation is blunt about ties: “The order in which two rows for which all ORDER BY expressions evaluate to equal values are returned is undefined.” A tiebreaker is an extra column, usually a unique id, that makes the order total.

Row value. SQLite’s name for a tuple such as (created_at, id). SQLite’s row value documentation explains the comparison rule: “Two row values are compared by looking at the constituent scalar values from left to right.”

Index. A sorted copy of some columns that lets the database find rows without reading the whole table. An index on several columns is sorted by the first, then the second, and so on.

Tenant. One customer account in a system that serves many. Every row belongs to one tenant, and a tenant must never see another tenant’s rows.

HMAC signature. A short tag computed from a message and a secret key. Anyone can read the message, but only someone who holds the key can produce a tag that checks out, so a changed message is detected.

Prerequisites

• Python 3.10 or newer. Everything here ran on Python 3.13.14 on Windows 11. The code uses the X | None type syntax, which needs 3.10.
• SQLite 3.15 or newer. Any current Python has it, and the check in the next section prints the version you have. Row values were added in that release; the documentation says “Row values were added to SQLite version 3.15.0 (2016-10-14).” Check yours in the next section.
• Basic SQL: SELECT, WHERE, ORDER BY and LIMIT. You do not need to know anything about indexes or FastAPI beforehand; both are explained when they appear.
• About 60 MB of free disk space for the lab database. There is no database server to install: SQLite runs inside Python.
• Optional: curl for the last step. On Windows PowerShell type curl.exe, because plain curl is an alias for a different command.

Create the project

Make a folder, create a virtual environment, activate it, and install four packages. The versions below are the ones this tutorial was run with. FastAPI’s test client in this version needs the httpx2 package, which is why it is in the list.

mkdir keyset
cd keyset
python -m venv .venv
.venv\Scripts\activate
python -m pip install fastapi==0.143.0 uvicorn==0.54.0 httpx2==2.13.1 pytest==9.1.1
python -c "import sqlite3; print(sqlite3.sqlite_version)"

On macOS and Linux the activation line is source .venv/bin/activate. The last command should print 3.50.4 or any version at or above 3.15.0. Every file below goes in the keyset folder and every command is run from there.

Step 1: Build a million-row table to practice on

Pagination problems only appear at scale, so you need a table that is big enough to hurt and messy enough to be realistic. Two small files do that.

1.1 The table and its index

Save this as db.py. It holds the schema and a helper that opens the database read-only, which is how the API will open it later.

"""db.py: schema and connection helpers shared by every step."""
import os
import sqlite3
from pathlib import Path

TABLE_SQL = """
CREATE TABLE IF NOT EXISTS events (
    id         INTEGER PRIMARY KEY,
    tenant_id  INTEGER NOT NULL,
    created_at INTEGER NOT NULL,
    kind       TEXT    NOT NULL,
    payload    TEXT    NOT NULL
);
"""
INDEX_SQL = (
    "CREATE INDEX IF NOT EXISTS idx_events_tenant_time "
    "ON events (tenant_id, created_at, id);"
)


def db_path() -> Path:
    return Path(os.environ.get("EVENTS_DB", Path(__file__).with_name("events.db")))


def connect_ro(path=None) -> sqlite3.Connection:
    """Open the database read-only, the way the API does."""
    uri = Path(path or db_path()).resolve().as_uri() + "?mode=ro"
    return sqlite3.connect(uri, uri=True)

Each row is an events record: an id that increases with each new event, the tenant_id of the customer it belongs to, a created_at time in whole seconds, and two descriptive columns. The index is the important part. It lists tenant_id first, then created_at, then id. That order is chosen to match the query you will write: one equality test on the tenant, then rows sorted by time and id. When the index order matches the sort order, the database can read rows straight out of the index in the right order and does not need a separate sorting step.

1.2 Fill it with one million events

Save this as step01_seed.py. It uses a fixed random seed, so you get the same rows every time. About 90 percent of the events belong to tenant 1; the other four tenants share the rest. About eight events happen every second across the whole system, so many events share a timestamp. Every 50,000 rows a “bulk import” writes 300 events for tenant 1 in the same second, which is the kind of burst real systems produce.

"""step01_seed.py: build events.db, one million rows, with timestamp ties and bursts."""
import random
import sqlite3
import time

from db import INDEX_SQL, TABLE_SQL, db_path

N = 1_000_000
START = 1_790_000_000                    # an arbitrary epoch second
PER_SECOND = 8                           # about eight events per second across all tenants
BURST_EVERY, BURST_SIZE = 50_000, 300    # a bulk import: 300 rows in the same second
KINDS = ["login", "logout", "upload", "delete", "update"]


def rows():
    rng = random.Random(7)
    burst_ts = 0
    for i in range(N):
        slot = i % BURST_EVERY
        if slot == 0:
            burst_ts = START + i // PER_SECOND
        if slot < BURST_SIZE:                       # the bulk import belongs to tenant 1
            ts, tenant = burst_ts, 1
        else:
            ts = START + i // PER_SECOND
            tenant = 1 if rng.random() < 0.9 else rng.randint(2, 5)
        yield i + 1, tenant, ts, rng.choice(KINDS), f"event-{i + 1}"


if __name__ == "__main__":
    path = db_path()
    path.unlink(missing_ok=True)
    t0 = time.perf_counter()
    conn = sqlite3.connect(path)
    conn.execute("PRAGMA journal_mode = OFF")
    conn.execute("PRAGMA synchronous = OFF")
    conn.executescript(TABLE_SQL)
    conn.executemany("INSERT INTO events VALUES (?, ?, ?, ?, ?)", rows())
    conn.executescript(INDEX_SQL)
    conn.commit()
    print(f"built {path.name} in {time.perf_counter() - t0:.1f} s, "
          f"{path.stat().st_size / 1e6:.0f} MB, SQLite {sqlite3.sqlite_version}")
    for tenant, n in conn.execute("SELECT tenant_id, COUNT(*) FROM events GROUP BY tenant_id"):
        print(f"  tenant {tenant}: {n:,} events")
    tied, biggest = conn.execute(
        "SELECT SUM(c), MAX(c) FROM (SELECT COUNT(*) AS c FROM events "
        "WHERE tenant_id = 1 GROUP BY created_at HAVING c > 1)").fetchone()
    print(f"  tenant 1: {tied:,} events share their second with another tenant 1 event; "
          f"the biggest tie group has {biggest} events")

Run it:

python step01_seed.py

built events.db in 1.2 s, 52 MB, SQLite 3.50.4
  tenant 1: 900,540 events
  tenant 2: 24,853 events
  tenant 3: 24,941 events
  tenant 4: 24,988 events
  tenant 5: 24,678 events
  tenant 1: 900,540 events share their second with another tenant 1 event; the biggest tie group has 300 events

Tenant 1 is the big customer, with 900,540 events. The last line is the one to remember: every one of those events shares its second with at least one other tenant 1 event, and the biggest group of identical timestamps has 300 events in it. Ties are not an edge case in this data. They are the normal case, and they are what breaks the first keyset attempt in Step 4.

Step 2: See what OFFSET gets wrong

Before building the fix, watch the problem happen. This script uses a tiny in-memory table of 12 events, newest first, with a page size of 5, so you can read every row id by eye. Save it as step02_drift.py.

"""step02_drift.py: what OFFSET paging does when rows arrive or disappear between two requests."""
import sqlite3

from db import INDEX_SQL, TABLE_SQL

PAGE = ("SELECT id FROM events WHERE tenant_id = 1 "
        "ORDER BY created_at DESC, id DESC LIMIT ? OFFSET ?")
NEW = "INSERT INTO events VALUES (?, 1, ?, 'login', ?)"


def fresh():
    conn = sqlite3.connect(":memory:")
    conn.executescript(TABLE_SQL + INDEX_SQL)
    conn.executemany(NEW, [(i, 1000 + i, f"event-{i}") for i in range(1, 13)])
    return conn


def ids(conn, limit, offset):
    return [row[0] for row in conn.execute(PAGE, (limit, offset))]


print("case 1: three new events arrive between the two requests")
conn = fresh()
page1 = ids(conn, 5, 0)
conn.executemany(NEW, [(i, 1000 + i, f"event-{i}") for i in (13, 14, 15)])
page2 = ids(conn, 5, 5)
print("  page 1:", page1)
print("  page 2:", page2)
print("  shown twice:", sorted(set(page1) & set(page2), reverse=True))

print("case 2: a cleanup job deletes two rows the client has already seen")
conn = fresh()
page1 = ids(conn, 5, 0)
conn.execute("DELETE FROM events WHERE id IN (11, 10)")
page2 = ids(conn, 5, 5)
print("  page 1:", page1)
print("  page 2:", page2)
print("  never shown:", sorted(set(range(1, 13)) - set(page1) - set(page2), reverse=True))

python step02_drift.py

case 1: three new events arrive between the two requests
  page 1: [12, 11, 10, 9, 8]
  page 2: [10, 9, 8, 7, 6]
  shown twice: [10, 9, 8]
case 2: a cleanup job deletes two rows the client has already seen
  page 1: [12, 11, 10, 9, 8]
  page 2: [5, 4, 3, 2, 1]
  never shown: [7, 6]

In the first case, three new events arrive between the two requests. They sort to the top, which pushes every older row three places down, so the first three rows of page 2 (10, 9 and 8) are the last three rows of page 1 again. The client shows them twice. In the second case a cleanup job deletes two rows the client had already read. Everything below them moves up two places, so page 2 starts two rows too late and rows 7 and 6 are never shown to anyone. Neither case is a bug in your code. The database did exactly what was asked: skip five rows. The trouble is that “five rows” stopped meaning “the five rows I already saw.”

Step 3: Measure what a deep OFFSET costs

Now the cost. This script asks tenant 1 for a page of 100 rows at deeper and deeper offsets and times each query, using the median of seven runs. It also prints the query plan, which is SQLite’s description of how it will run a query. Save it as step03_cost.py.

"""step03_cost.py: time OFFSET pages at growing depth, and show the plan."""
import statistics
import time

from db import connect_ro

SQL = ("SELECT id, created_at, kind, payload FROM events WHERE tenant_id = 1 "
       "ORDER BY created_at DESC, id DESC LIMIT 100 OFFSET ?")


def median_ms(conn, offset, repeats=7):
    times = []
    for _ in range(repeats):
        t0 = time.perf_counter()
        conn.execute(SQL, (offset,)).fetchall()
        times.append((time.perf_counter() - t0) * 1000)
    return statistics.median(times)


conn = connect_ro()
total = conn.execute("SELECT COUNT(*) FROM events WHERE tenant_id = 1").fetchone()[0]
print(f"tenant 1 has {total:,} events, page size 100")
for row in conn.execute("EXPLAIN QUERY PLAN " + SQL, (0,)):
    print("plan:", row[3])
median_ms(conn, 0)                                    # warm the cache once
print(f"\n{'offset':>9}  {'median ms':>9}")
for offset in (0, 1_000, 10_000, 100_000, 400_000, 800_000):
    print(f"{offset:>9,}  {median_ms(conn, offset):>9.2f}")

python step03_cost.py

tenant 1 has 900,540 events, page size 100
plan: SEARCH events USING INDEX idx_events_tenant_time (tenant_id=?)

   offset  median ms
        0       0.09
    1,000       0.11
   10,000       0.27
  100,000       1.82
  400,000      11.25
  800,000      22.89

The plan line says SEARCH events USING INDEX idx_events_tenant_time (tenant_id=?): the database uses the index, reads rows already in order, and has no separate sorting step. That is the best case, and the cost still climbs from 0.09 milliseconds at offset 0 to about 23 milliseconds at offset 800,000, roughly 250 times slower for the same 100 rows. The SQLite documentation describes the same effect in its section on scrolling windows: “OFFSET requires time proportional to the offset value.” Twenty-three milliseconds sounds small until you remember it is CPU time spent on every request that reaches that depth. At about 40 such requests a second one core is fully busy producing rows to discard, and if the client chooses the offset, anyone can send those requests. The backpressure and load shedding tutorial covers what to do when a service is already saturated; the better move is to not offer an endpoint whose cost the caller controls.

Step 4: A first keyset attempt, and why it silently loses rows

The idea of keyset pagination is simple. Instead of counting rows, remember where you stopped. The server returns the sort key of the last row on the page, the client sends it back, and the next query asks for rows that sort after it. Since the data is sorted by created_at, the obvious first attempt is WHERE created_at < ? with the last time you saw. Save this as step04_ties.py. It walks all of tenant 1 twice, once with < and once with <=, and counts what comes back.

"""step04_ties.py: keyset paging on created_at alone, first with < and then with <=."""
from db import connect_ro

LIMIT = 100
FIRST = ("SELECT id, created_at FROM events WHERE tenant_id = 1 "
         "ORDER BY created_at DESC LIMIT ?")
AFTER_LT = ("SELECT id, created_at FROM events WHERE tenant_id = 1 AND created_at < ? "
            "ORDER BY created_at DESC LIMIT ?")
AFTER_LE = ("SELECT id, created_at FROM events WHERE tenant_id = 1 AND created_at <= ? "
            "ORDER BY created_at DESC LIMIT ?")


def walk(conn, after_sql):
    ids, pages = [], 0
    rows = conn.execute(FIRST, (LIMIT,)).fetchall()
    while rows:
        pages += 1
        ids += [row[0] for row in rows]
        nxt = conn.execute(after_sql, (rows[-1][1], LIMIT)).fetchall()
        if nxt and nxt[-1] == rows[-1]:               # the same last row twice: no progress
            return ids, pages, rows[-1][1]
        rows = nxt
    return ids, pages, None


conn = connect_ro()
total = conn.execute("SELECT COUNT(*) FROM events WHERE tenant_id = 1").fetchone()[0]

ids, pages, _ = walk(conn, AFTER_LT)
missing = total - len(set(ids))
print("cursor test: created_at < ?")
print(f"  {pages:,} pages, {len(ids):,} rows returned, {len(ids) - len(set(ids))} repeated")
print(f"  {missing:,} of {total:,} rows ({missing / total:.1%}) were never returned")

ids, pages, stuck_at = walk(conn, AFTER_LE)
print("cursor test: created_at <= ?")
print(f"  stuck after page {pages:,}: created_at={stuck_at} returns the same page again and again")
print(f"  {len(ids) - len(set(ids))} rows had already been returned twice by then")

python step04_ties.py

cursor test: created_at < ?
  8,687 pages, 868,700 rows returned, 0 repeated
  31,840 of 900,540 rows (3.5%) were never returned
cursor test: created_at <= ?
  stuck after page 468: created_at=1790118750 returns the same page again and again
  2090 rows had already been returned twice by then

With <, the walk finishes cleanly, returns no row twice, and still loses 31,840 of 900,540 rows, 3.5 percent of the table, with no error and no warning. With <=, the walk gets stuck on page 468, where every further request returns the same page. Both failures have one cause: ties. Suppose a page ends on a row with created_at equal to T, and four more rows with the same T have not been returned yet. The condition created_at < T skips those four, because they are not smaller than T. The condition created_at <= T returns them, but it also returns every row with time T that you already have, and when a tie group is larger than the page size, as the 300-event bulk import is, once a page holds only rows from that group every following page is the same and the cursor never moves.

The rule that comes out of this: a keyset sort key must be unique. When the natural sort column has duplicates, add a column that makes each row distinct, here the id, and sort by both. SQLite’s statement about ties, quoted in the terms section above, is why you must say so explicitly: without a tiebreaker the order of tied rows is undefined, and a cursor needs a defined order to point into.

Step 5: Fix it with a tiebreaker and a row value

With the order created_at DESC, id DESC, no two rows sort equal, so “the rows after this one” means exactly one thing: rows whose (created_at, id) pair is smaller than the last row’s pair. SQLite can compare pairs directly with a row value, and the comparison runs from left to right: the first column decides, and the second is consulted only on a tie. The SQLite documentation recommends this very pattern: “A more efficient approach is to remember the last entry currently displayed and then use a row value comparison in the WHERE clause”. Save this as pager.py.

"""pager.py: keyset ("seek") pagination for the events table."""
from typing import NamedTuple


class Position(NamedTuple):
    """Where the previous page ended: the sort key of its last row."""
    created_at: int
    id: int


def build_query(kind: bool, after: bool) -> str:
    """Assemble the page query from fixed fragments. Values only travel as ? parameters."""
    parts = ["SELECT id, created_at, kind, payload FROM events WHERE tenant_id = ?"]
    if kind:
        parts.append("AND kind = ?")
    if after:
        parts.append("AND (created_at, id) < (?, ?)")
    parts.append("ORDER BY created_at DESC, id DESC LIMIT ?")
    return " ".join(parts)


def fetch_page(conn, tenant_id, limit, after=None, kind=None):
    """Return (rows, next_position). next_position is None on the last page."""
    args = [tenant_id]
    if kind is not None:
        args.append(kind)
    if after is not None:
        args += [after.created_at, after.id]
    args.append(limit + 1)               # one extra row tells us whether another page exists
    rows = conn.execute(build_query(kind is not None, after is not None), args).fetchall()
    page = rows[:limit]
    more = len(rows) > limit
    return page, (Position(page[-1][1], page[-1][0]) if more else None)


def walk_all(conn, tenant_id, limit, kind=None):
    """Follow the pages to the end and return every id in the order delivered."""
    after, ids = None, []
    while True:
        page, after = fetch_page(conn, tenant_id, limit, after, kind)
        ids += [row[0] for row in page]
        if after is None:
            return ids

Read it from the top. Position is the pair of values that identifies where a page ended. build_query assembles the SQL from fixed text fragments, adding the optional kind filter and the cursor condition only when they are needed. No value from a client is ever pasted into the SQL text; values travel only as ? parameters, which is the defense explained in the SQL injection tutorial. fetch_page asks for limit + 1 rows. If the extra row exists, there is another page and the last row of the page the client gets defines the next position; if it does not exist, the position is None and the walk is over. That extra row costs almost nothing and removes the need for a separate count query.

Now check it. Save step05_walk.py. Its first half walks all of tenant 1 in pages of 100. Its second half repeats the “table keeps changing” experiment on a 1,000-row table, with new events arriving and old rows being deleted between requests, once with OFFSET and once with keyset.

"""step05_walk.py: keyset paging with a tie-breaker, on the full table and on a table that keeps changing."""
import random
import sqlite3
import time

from db import INDEX_SQL, TABLE_SQL, connect_ro
from pager import fetch_page, walk_all

# 1. Walk every event of tenant 1, 100 per page.
conn = connect_ro()
want = {row[0] for row in conn.execute("SELECT id FROM events WHERE tenant_id = 1")}
t0 = time.perf_counter()
got = walk_all(conn, 1, 100)
secs = time.perf_counter() - t0
print("full walk of tenant 1, 100 rows per page")
print(f"  {len(got):,} rows returned, {len(got) - len(set(got))} repeated, "
      f"{len(want - set(got))} missing ({secs:.1f} s)")

# step05_walk.py (continued)
# 2. The same walk while the table changes between requests.
OFFSET_PAGE = ("SELECT id FROM events WHERE tenant_id = 1 "
               "ORDER BY created_at DESC, id DESC LIMIT ? OFFSET ?")
NEW = "INSERT INTO events VALUES (?, 1, ?, 'login', ?)"


def small_table():
    db = sqlite3.connect(":memory:")
    db.executescript(TABLE_SQL + INDEX_SQL)
    db.executemany(NEW, [(i, 5000 + i // 4, f"event-{i}") for i in range(1, 1001)])
    return db


def churn_walk(strategy, seed=3):
    rng = random.Random(seed)
    db, delivered, deleted = small_table(), [], set()
    after, offset, next_id, pages = None, 0, 1001, 0
    while True:
        if strategy == "offset":
            page = [row[0] for row in db.execute(OFFSET_PAGE, (7, offset))]
            offset += 7
            finished = not page
        else:
            rows, after = fetch_page(db, 1, 7, after)
            page, finished = [row[0] for row in rows], after is None
        delivered += page
        pages += 1
        if finished:
            break
        for _ in range(rng.randint(0, 3)):               # new events arrive
            db.execute(NEW, (next_id, 5000 + next_id // 4, f"event-{next_id}"))
            next_id += 1
        for _ in range(rng.randint(0, 2)):               # rows the client has seen are deleted
            victim = rng.choice(delivered)
            db.execute("DELETE FROM events WHERE id = ?", (victim,))
            deleted.add(victim)
        ahead = sorted(set(range(1, 1001)) - set(delivered) - deleted)
        if ahead and rng.random() < 0.5:                 # sometimes a row it has not seen yet
            victim = rng.choice(ahead)
            db.execute("DELETE FROM events WHERE id = ?", (victim,))
            deleted.add(victim)
    repeated = len(delivered) - len(set(delivered))
    missed = len(set(range(1, 1001)) - deleted - set(delivered))
    return pages, len(delivered), repeated, missed


print("\nwalk of a 1,000-row table that changes between requests, 7 rows per page")
print(f"{'strategy':<8} {'pages':>5} {'delivered':>9} {'repeated':>8} {'missed':>6}")
for strategy in ("offset", "keyset"):
    pages, delivered, repeated, missed = churn_walk(strategy)
    print(f"{strategy:<8} {pages:>5} {delivered:>9} {repeated:>8} {missed:>6}")

python step05_walk.py

full walk of tenant 1, 100 rows per page
  900,540 rows returned, 0 repeated, 0 missing (1.0 s)

walk of a 1,000-row table that changes between requests, 7 rows per page
strategy pages delivered repeated missed
offset     148      1027      131     37
keyset     134       937        0      0

The full walk returns all 900,540 rows with none repeated and none missing, in about a second. The comparison table is the more interesting result. The OFFSET walk returned 131 rows a second time and never returned 37 rows that sat in the table the whole time. The keyset walk returned 937 rows, none twice, and “missed 0” means that every row it did not return had been deleted before the walk reached it. New events landed above the cursor, so a walk already under way never saw them, and the next walk, starting from the top, will. That is the behavior you want: a stable window moving through the data, not a numbered list that shifts under the reader.

Step 6: Read the query plan before you trust the speed

A query can be correct and still slow, and the way to tell is the plan. EXPLAIN QUERY PLAN in front of any SQLite query prints how it will run. The SQLite EXPLAIN QUERY PLAN tutorial goes through the vocabulary in detail. The short version for this step is that SEARCH ... USING INDEX with a condition in brackets means the database jumps straight into the index at the condition, and a plan that mentions only the tenant means it starts from the newest row and filters as it goes. Save step06_plans.py. It prints the plan and times the page that follows the cursor at depth 800,000 for five ways of writing the same request, and, except for the filtered one, checks that each returns the same 100 rows as the OFFSET query.

"""step06_plans.py: read the query plans, then time the deepest page for each way to write the cursor test."""
import statistics
import time

from db import connect_ro
from pager import Position, build_query

conn = connect_ro()
pos = Position(*conn.execute(
    "SELECT created_at, id FROM events WHERE tenant_id = 1 "
    "ORDER BY created_at DESC, id DESC LIMIT 1 OFFSET 799999").fetchone())

HEAD = "SELECT id, created_at, kind, payload FROM events WHERE tenant_id = ?"
TAIL = "ORDER BY created_at DESC, id DESC LIMIT ?"
QUERIES = {
    "OFFSET 800000": (HEAD + " ORDER BY created_at DESC, id DESC LIMIT ? OFFSET ?",
                      (1, 100, 800_000)),
    "row value": (build_query(kind=False, after=True),
                  (1, pos.created_at, pos.id, 100)),
    "expanded OR": (HEAD + " AND (created_at < ? OR (created_at = ? AND id < ?)) " + TAIL,
                    (1, pos.created_at, pos.created_at, pos.id, 100)),
    "range + expanded OR": (HEAD + " AND created_at <= ? AND (created_at < ? OR id < ?) " + TAIL,
                            (1, pos.created_at, pos.created_at, pos.id, 100)),
    "row value + kind filter": (build_query(kind=True, after=True),
                                (1, "upload", pos.created_at, pos.id, 100)),
}


def median_ms(sql, args, repeats=9):
    conn.execute(sql, args).fetchall()                # warm up
    times = []
    for _ in range(repeats):
        t0 = time.perf_counter()
        conn.execute(sql, args).fetchall()
        times.append((time.perf_counter() - t0) * 1000)
    return statistics.median(times)


reference = [row[0] for row in conn.execute(*QUERIES["OFFSET 800000"])]
for name, (sql, args) in QUERIES.items():
    print(name)
    for row in conn.execute("EXPLAIN QUERY PLAN " + sql, args):
        print("   plan:", row[3])
    print(f"   page after the same cursor: {median_ms(sql, args):.2f} ms")
    if "kind" not in name:
        same = [row[0] for row in conn.execute(sql, args)] == reference
        print(f"   same 100 rows as the OFFSET page: {same}")

python step06_plans.py

OFFSET 800000
   plan: SEARCH events USING INDEX idx_events_tenant_time (tenant_id=?)
   page after the same cursor: 23.12 ms
   same 100 rows as the OFFSET page: True
row value
   plan: SEARCH events USING INDEX idx_events_tenant_time (tenant_id=? AND created_at<?)
   page after the same cursor: 0.10 ms
   same 100 rows as the OFFSET page: True
expanded OR
   plan: SEARCH events USING INDEX idx_events_tenant_time (tenant_id=?)
   page after the same cursor: 32.82 ms
   same 100 rows as the OFFSET page: True
range + expanded OR
   plan: SEARCH events USING INDEX idx_events_tenant_time (tenant_id=? AND created_at<?)
   page after the same cursor: 0.10 ms
   same 100 rows as the OFFSET page: True
row value + kind filter
   plan: SEARCH events USING INDEX idx_events_tenant_time (tenant_id=? AND created_at<?)
   page after the same cursor: 0.16 ms

All of the first four versions return the same 100 rows, so they are equally correct. They are not equally fast. The row value comparison from pager.py takes about 0.1 milliseconds, and its plan shows created_at<? inside the brackets: the database seeks into the index instead of walking it. The expanded form, created_at < ? OR (created_at = ? AND id < ?), says the same thing in logic but its plan mentions only the tenant, and it takes about 33 milliseconds, slower even than the OFFSET query. In this SQLite version, with this index, the planner does not turn that OR into a range search, so it starts at the newest row and tests the condition on rows until it reaches the cursor. Adding a redundant range, created_at <= ?, in front of the OR gives it something to seek on and brings the time back to about 0.1 milliseconds. You will need that trick in Step 10. The last entry shows the kind filter on top of the row value: still a seek, at about 0.16 milliseconds. Whatever form you write, look at the plan.

Step 7: Turn the position into a signed, opaque cursor

You could now return created_at and id to the client and ask for them back. Many APIs start there. Do not ship it that way, for three reasons, and Google’s API Improvement Proposal 158 on pagination backs each of them. First, a cursor should be opaque: “Page tokens provided by APIs must be opaque (but URL-safe) strings, and must not be user-parseable.” If clients can read the position, they will build on it, and your sort order becomes part of your API, and changing it breaks them. Second, base64 does not count as hiding anything: “Base-64 encoding an otherwise-transparent page token is not a sufficient obfuscation mechanism.” Anyone can decode and edit it, which is exactly what Step 9 does. Third, a cursor belongs to one query. The same proposal expects clients to “keep all other arguments to the RPC the same; if any arguments are different, the API should send an INVALID_ARGUMENT error.” A cursor made for kind=login and used with kind=upload would silently return rows from the wrong place.

The design that follows from this has three parts. The token carries the position, a version number, and a fingerprint, a short hash of everything that shapes the result set (the tenant and the filters). It is signed with an HMAC, so any edit is detected. And the signature is checked before anything inside the token is parsed. Save this as cursor.py, in two parts.

"""cursor.py: turn a Position into an opaque, signed token, and back."""
import base64
import hashlib
import hmac
import json

from pager import Position


class BadCursor(ValueError):
    """The token is not one this server issued for this query."""


def _b64e(raw: bytes) -> str:
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")


def _b64d(text: str) -> bytes:
    return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))


def fingerprint(**query) -> str:
    """A short, stable hash of everything that shapes the result set (tenant, filters)."""
    blob = json.dumps(query, sort_keys=True, separators=(",", ":")).encode()
    return hashlib.sha256(blob).hexdigest()[:16]

# cursor.py (continued)
def encode(pos: Position, fp: str, key: bytes) -> str:
    body = json.dumps({"v": 1, "c": pos.created_at, "i": pos.id, "f": fp},
                      separators=(",", ":")).encode()
    tag = hmac.new(key, body, hashlib.sha256).digest()
    return _b64e(body) + "." + _b64e(tag)


def decode(token: str, fp: str, key: bytes) -> Position:
    try:
        body_part, tag_part = token.split(".")
        body, tag = _b64d(body_part), _b64d(tag_part)
    except ValueError:                    # wrong number of dots, bad base64, non-ASCII text
        raise BadCursor("malformed cursor") from None
    expected = hmac.new(key, body, hashlib.sha256).digest()
    if not hmac.compare_digest(tag, expected):
        raise BadCursor("bad signature")  # nothing from the body is parsed before this check
    try:
        data = json.loads(body)
        ok = (data["v"] == 1 and type(data["c"]) is int
              and type(data["i"]) is int and data["f"] == fp)
    except (ValueError, KeyError, TypeError):
        raise BadCursor("malformed cursor") from None
    if not ok:
        raise BadCursor("cursor does not match this query")
    return Position(data["c"], data["i"])

The token has the form body.tag. The body is a small JSON object encoded with URL-safe base64, which turns bytes into letters, digits, hyphen and underscore so the token can sit in a query string. The tag is the HMAC-SHA256 of the body under a secret key. decode splits the token, recomputes the tag, and compares the two with hmac.compare_digest. Python’s hmac documentation explains why a plain == is the wrong tool: “This function uses an approach designed to prevent timing analysis by avoiding content-based short circuiting behaviour, making it appropriate for cryptography.” Every way a token can be wrong ends in the same BadCursor exception, so the endpoint can turn it into one clean error. The v field is there for the day you change the format: bump it and old tokens are refused instead of misread. Try it with step07_cursor.py:

"""step07_cursor.py: what a cursor looks like, and which check fires for each kind of bad token."""
import base64
import hashlib
import hmac

from cursor import BadCursor, _b64e, decode, encode, fingerprint
from pager import Position

KEY = b"demo-key-not-for-production"
fp = fingerprint(tenant=2, kind=None)
pos = Position(1_790_100_000, 424_242)
token = encode(pos, fp, KEY)
body_part, tag_part = token.split(".")

print("token  :", token)
print("length :", len(token), "characters")
print("body   :", base64.urlsafe_b64decode(body_part + "=" * (-len(body_part) % 4)).decode())
print("decoded:", decode(token, fp, KEY))
print()


def attempt(label, tok, query_fp=fp, key=KEY):
    try:
        decode(tok, query_fp, key)
        print(f"{label:<44} accepted")
    except BadCursor as exc:
        print(f"{label:<44} rejected: {exc}")


edited = encode(Position(1_790_100_000, 1), fp, KEY).split(".")[0]      # a body with id 1
other_filter = fingerprint(tenant=2, kind="login")
signed_junk_body = b"[1, 2]"
signed_junk = _b64e(signed_junk_body) + "." + _b64e(
    hmac.new(KEY, signed_junk_body, hashlib.sha256).digest())

attempt("the token we issued", token)
attempt("body edited to id 1, old signature", edited + "." + tag_part)
attempt("signature changed", body_part + "." + _b64e(b"x" * 32))
attempt("signed with a different key", token, key=b"another-key")
attempt("same token, different filter (kind=login)", token, query_fp=other_filter)
attempt("not a cursor at all", "garbage")
attempt("valid signature, body is not an object", signed_junk)

python step07_cursor.py

token  : eyJ2IjoxLCJjIjoxNzkwMTAwMDAwLCJpIjo0MjQyNDIsImYiOiI3Y2NkMjFiMTcyMTEwNzkwIn0.lpVsPMGKY7RG-G4ZCt-cgodvUiFZ32T1hqKTQmzVilU
length : 119 characters
body   : {"v":1,"c":1790100000,"i":424242,"f":"7ccd21b172110790"}
decoded: Position(created_at=1790100000, id=424242)

the token we issued                          accepted
body edited to id 1, old signature           rejected: bad signature
signature changed                            rejected: bad signature
signed with a different key                  rejected: bad signature
same token, different filter (kind=login)    rejected: cursor does not match this query
not a cursor at all                          rejected: malformed cursor
valid signature, body is not an object       rejected: malformed cursor

The token is 119 characters. The body decodes to a plain JSON object, so this token is signed, not hidden: a client can read the position but cannot change it. AIP-158 asks for tokens that users cannot parse. If the position values are sensitive for your data, encrypt the body with an authenticated cipher, or keep the position on the server and hand out a random identifier instead. Here the values are a timestamp and a row id, so a signature is enough, and it is what stops the attack in Step 9. Each rejection comes from a different check. Editing the body or the tag, or signing with another key, fails the signature. Using the token for a different filter passes the signature but fails the fingerprint comparison. Garbage fails when splitting or decoding. A token with a valid signature but a body that is not the right shape fails the last check, which matters because in a real system a key leak or a bug in your own code could produce such a token. In production, read the key from an environment variable or a secrets manager, make it long and random, for example with python -c "import secrets; print(secrets.token_urlsafe(32))", and never put it in the repository.

Step 8: Build the endpoint

Now connect the pieces to FastAPI, a Python web framework that turns ordinary functions into HTTP endpoints and validates the query parameters and headers declared in their arguments. Save this as app.py. Read the three decisions in it before running anything.

"""app.py: GET /events with signed keyset cursors."""
import os
import sqlite3
from pathlib import Path
from typing import Literal

from fastapi import Depends, FastAPI, Header, HTTPException, Query

from cursor import BadCursor, decode, encode, fingerprint
from db import db_path
from pager import fetch_page

CURSOR_KEY = os.environ.get("CURSOR_KEY", "dev-only-change-me").encode()
API_KEYS = {"key-tenant-1": 1, "key-tenant-2": 2}       # stand-in for real authentication

app = FastAPI()


def tenant_id(x_api_key: str | None = Header(default=None)) -> int:
    """The tenant comes from the credential, never from the cursor or the query string."""
    if x_api_key not in API_KEYS:
        raise HTTPException(status_code=401, detail="invalid API key")
    return API_KEYS[x_api_key]


def get_db():
    uri = Path(db_path()).resolve().as_uri() + "?mode=ro"
    # one connection per request, used by one thread at a time
    conn = sqlite3.connect(uri, uri=True, check_same_thread=False)
    try:
        yield conn
    finally:
        conn.close()


def to_item(row):
    return {"id": row[0], "created_at": row[1], "kind": row[2], "payload": row[3]}


@app.get("/events")
def list_events(
    limit: int = Query(50, ge=1, le=100),
    cursor: str | None = None,
    kind: Literal["login", "logout", "upload", "delete", "update"] | None = None,
    tenant: int = Depends(tenant_id),
    db: sqlite3.Connection = Depends(get_db),
):
    fp = fingerprint(tenant=tenant, kind=kind)
    after = None
    if cursor is not None:
        try:
            after = decode(cursor, fp, CURSOR_KEY)
        except BadCursor:
            raise HTTPException(status_code=400, detail="invalid cursor") from None
    rows, nxt = fetch_page(db, tenant, limit, after, kind)
    return {
        "items": [to_item(row) for row in rows],
        "next_cursor": encode(nxt, fp, CURSOR_KEY) if nxt else None,
    }

The first decision is where the tenant comes from. tenant_id is a FastAPI dependency, a function that FastAPI runs before the endpoint and whose result it hands to the endpoint. It reads the X-API-Key header and looks it up in a dictionary, which stands in for real authentication such as a session or a token; the point is that the tenant comes from the credential and never from the cursor or the query string. The same Google proposal says why: page tokens “must not provide any form of authorization to the underlying resources, and authorization must be performed on the request as with any other regardless of the presence of a page token.” The broken object level authorization tutorial shows what goes wrong when it is the other way around. The second decision is the connection. get_db opens the database read-only with mode=ro, so a bug in the list endpoint cannot write. Python’s sqlite3 module normally refuses to use a connection from a different thread than the one that created it, and FastAPI may run the dependency and the endpoint on different worker threads, so the code turns that check off; this is safe because every request gets its own connection and uses it one step at a time. The third decision is the limit. Query(50, ge=1, le=100) gives a default of 50 and rejects anything below 1 or above 100 with a 422 error before any SQL runs. Stripe’s API, which also uses cursors, accepts a limit “ranging between 1 and 100”, and its pagination page says the list methods “use cursor-based pagination through the starting_after and ending_before parameters.” Stripe’s cursor is an object ID. When the sort key has ties, as the timestamps in this table do, the pair you built above is what pins down a position.

Now drive it. step08_api.py uses FastAPI’s test client, which calls the app directly without starting a server. It fetches a first page, walks every page of one tenant with and without a filter, and then sends six bad requests.

"""step08_api.py: drive the endpoint with a test client."""
import json

from fastapi.testclient import TestClient

from app import app
from db import connect_ro

client = TestClient(app)
H1, H2 = {"X-API-Key": "key-tenant-1"}, {"X-API-Key": "key-tenant-2"}


def walk(headers, **params):
    """Follow next_cursor until it is null; return (ids, pages)."""
    ids, pages, cursor = [], 0, None
    while True:
        query = dict(params, limit=100)
        if cursor:
            query["cursor"] = cursor
        body = client.get("/events", params=query, headers=headers).json()
        ids += [item["id"] for item in body["items"]]
        pages += 1
        cursor = body["next_cursor"]
        if cursor is None:
            return ids, pages


r = client.get("/events", params={"limit": 2}, headers=H2)
print(r.status_code, json.dumps(r.json(), indent=2))

db = connect_ro()
for label, params in (("tenant 2, all kinds", {}), ("tenant 2, kind=upload", {"kind": "upload"})):
    ids, pages = walk(H2, **params)
    if "kind" in params:
        total = db.execute("SELECT COUNT(*) FROM events WHERE tenant_id = 2 AND kind = ?",
                           (params["kind"],)).fetchone()[0]
    else:
        total = db.execute("SELECT COUNT(*) FROM events WHERE tenant_id = 2").fetchone()[0]
    print(f"\n{label}: {pages} pages, {len(ids):,} rows returned, "
          f"{len(ids) - len(set(ids))} repeated, table holds {total:,}")

good = client.get("/events", params={"limit": 2}, headers=H2).json()["next_cursor"]
login = client.get("/events", params={"limit": 2, "kind": "login"}, headers=H2).json()["next_cursor"]
body_part, tag_part = good.split(".")
flipped = ("A" if body_part[0] != "A" else "B") + body_part[1:] + "." + tag_part

cases = [
    ("cursor with one character changed", dict(params={"cursor": flipped}, headers=H2)),
    ("kind=login cursor reused for kind=upload", dict(params={"cursor": login, "kind": "upload"}, headers=H2)),
    ("tenant 2 cursor sent with tenant 1 key", dict(params={"cursor": good}, headers=H1)),
    ("garbage cursor", dict(params={"cursor": "garbage"}, headers=H2)),
    ("limit=1000", dict(params={"limit": 1000}, headers=H2)),
    ("no API key", dict(params={})),
]
print()
for label, kwargs in cases:
    r = client.get("/events", **kwargs)
    detail = r.json()["detail"]
    if isinstance(detail, list):
        detail = detail[0]["msg"]
    print(f"{r.status_code}  {label:<42} {detail}")

python step08_api.py

200 {
  "items": [
    {
      "id": 999955,
      "created_at": 1790124994,
      "kind": "delete",
      "payload": "event-999955"
    },
    {
      "id": 999948,
      "created_at": 1790124993,
      "kind": "update",
      "payload": "event-999948"
    }
  ],
  "next_cursor": "eyJ2IjoxLCJjIjoxNzkwMTI0OTkzLCJpIjo5OTk5NDgsImYiOiI3Y2NkMjFiMTcyMTEwNzkwIn0.Xf9PUGj7T7x-zq1N9CfKOH1B9yjvBXGoFzvb1jkOHtw"
}

The response has two fields. items is the page, newest first. next_cursor is the token to send back as ?cursor=, and it is null on the last page, the only signal the client needs to stop. The walks come next:

tenant 2, all kinds: 249 pages, 24,853 rows returned, 0 repeated, table holds 24,853

tenant 2, kind=upload: 51 pages, 5,033 rows returned, 0 repeated, table holds 5,033

Tenant 2 owns 24,853 events, and 249 requests of up to 100 rows returned every one of them exactly once; the filtered walk returned all 5,033 uploads. The six bad requests show the error handling:

400  cursor with one character changed          invalid cursor
400  kind=login cursor reused for kind=upload   invalid cursor
400  tenant 2 cursor sent with tenant 1 key     invalid cursor
400  garbage cursor                             invalid cursor
422  limit=1000                                 Input should be less than or equal to 100
401  no API key                                 invalid API key

A changed cursor, a cursor reused with a different filter, a cursor sent with a different tenant’s key, and plain garbage all get the same 400 invalid cursor, and the message is deliberately the same for all four so the response does not tell an attacker which check failed. An out-of-range limit gets FastAPI’s own 422 validation message, and a missing key gets 401.

Step 9: See why the signature matters

It helps to watch the unsigned version fail. app_unsigned.py is the tempting design: the cursor is just the query state, tenant included, base64 encoded, and the server trusts whatever it decodes. Save it, and then save step09_attack.py, which plays a client of tenant 2 who decodes its own cursor and edits it.

"""app_unsigned.py: the tempting version, where the cursor is just the query state, base64 encoded."""
import base64
import json

from fastapi import Depends, FastAPI

from app import get_db, tenant_id, to_item
from pager import Position, fetch_page

app = FastAPI()


def to_token(state: dict) -> str:
    return base64.urlsafe_b64encode(json.dumps(state).encode()).decode()


def from_token(token: str) -> dict:
    return json.loads(base64.urlsafe_b64decode(token))


@app.get("/events")
def list_events(limit: int = 50, cursor: str | None = None,
                tenant: int = Depends(tenant_id), db=Depends(get_db)):
    state = from_token(cursor) if cursor else {"tenant": tenant, "after": None}
    after = Position(*state["after"]) if state["after"] else None
    rows, nxt = fetch_page(db, state["tenant"], limit, after)     # the cursor decides whose rows
    next_state = {"tenant": state["tenant"], "after": list(nxt) if nxt else None}
    return {"items": [to_item(row) for row in rows],
            "next_cursor": to_token(next_state) if nxt else None}

"""step09_attack.py: the cursor that carries tenant state, and what a client can do with it."""
from fastapi.testclient import TestClient

import app as signed
import app_unsigned as unsigned
from db import connect_ro

H2 = {"X-API-Key": "key-tenant-2"}
db = connect_ro()


def owners(ids):
    return sorted({db.execute("SELECT tenant_id FROM events WHERE id = ?", (i,)).fetchone()[0]
                   for i in ids})


bad = TestClient(unsigned.app)
page1 = bad.get("/events", params={"limit": 3}, headers=H2).json()
print("tenant 2 asks for 3 events:", [item["id"] for item in page1["items"]])
state = unsigned.from_token(page1["next_cursor"])
print("the cursor decodes to:", state)
state["tenant"] = 1
forged = unsigned.to_token(state)
leak = bad.get("/events", params={"limit": 5, "cursor": forged}, headers=H2).json()
ids = [item["id"] for item in leak["items"]]
print(f"forged cursor: {len(ids)} events returned {ids}, owned by tenant(s) {owners(ids)}")

good = TestClient(signed.app)
page1 = good.get("/events", params={"limit": 3}, headers=H2).json()
body_part, tag_part = page1["next_cursor"].split(".")
other = unsigned.to_token({"v": 1, "c": 1_790_000_000, "i": 1, "f": "x"}).rstrip("=")
r = good.get("/events", params={"limit": 5, "cursor": other + "." + tag_part}, headers=H2)
print(f"\nsigned version, edited body with the old signature: {r.status_code} {r.json()['detail']}")

python step09_attack.py

tenant 2 asks for 3 events: [999955, 999948, 999921]
the cursor decodes to: {'tenant': 2, 'after': [1790124990, 999921]}
forged cursor: 5 events returned [999920, 999919, 999918, 999917, 999916], owned by tenant(s) [1]

signed version, edited body with the old signature: 400 invalid cursor

The client of tenant 2 asked for three events and got a cursor that decodes to {'tenant': 2, 'after': [...]}. It changed the tenant to 1, encoded it again, and sent it with its own key. The server obeyed the cursor, and tenant 2 received five events owned by tenant 1. That is a broken object level authorization bug that no authentication check would catch, because the caller really is tenant 2. The signed version of the endpoint, shown on the last line, refuses a body that was edited while the old signature was kept, because the attacker has no way to compute a new one. Two things protect you, and you need both: the tenant comes from the credential, and the cursor cannot be forged without the signing key.

Step 10: Three traps that remain

A working endpoint can still fail in ways the happy path never shows. Save step10_gotchas.py. It reproduces the first two traps on tiny tables; the third is a limit, not a bug.

"""step10_gotchas.py: two ways a keyset query goes quietly wrong."""
import sqlite3

# 1. A nullable sort column. NULL compared with anything is never true.
FIRST = "SELECT id, due_at FROM tickets ORDER BY due_at DESC, id DESC LIMIT 4"
NEXT = ("SELECT id, due_at FROM tickets WHERE (due_at, id) < (?, ?) "
        "ORDER BY due_at DESC, id DESC LIMIT 4")


def tickets(no_due_date):
    db = sqlite3.connect(":memory:")
    db.execute("CREATE TABLE tickets (id INTEGER PRIMARY KEY, due_at INTEGER)")
    db.executemany("INSERT INTO tickets VALUES (?, ?)",
                   [(i, 100 + i if i <= 7 else no_due_date) for i in range(1, 11)])
    return db


def walk(db):
    seen, rows = [], db.execute(FIRST).fetchall()
    while rows:
        seen += [row[0] for row in rows]
        rows = db.execute(NEXT, (rows[-1][1], rows[-1][0])).fetchall()
    return seen


for label, value in (("due_at is NULL when unknown", None), ("due_at is 0 when unknown", 0)):
    seen = walk(tickets(value))
    print(f"{label}: {len(seen)} of 10 rows returned, never returned {sorted(set(range(1, 11)) - set(seen))}")

# step10_gotchas.py (continued)
# 2. Mixed sort directions. A row value compares every column the same way.
ROW_VALUE = ("SELECT id, created_at FROM events WHERE (created_at, id) < (?, ?) "
             "ORDER BY created_at DESC, id ASC LIMIT 4")
EXPANDED = ("SELECT id, created_at FROM events "
            "WHERE created_at < ? OR (created_at = ? AND id > ?) "
            "ORDER BY created_at DESC, id ASC LIMIT 4")
db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE events (id INTEGER PRIMARY KEY, created_at INTEGER NOT NULL)")
db.executemany("INSERT INTO events VALUES (?, ?)", [(i, 100 + i // 3) for i in range(1, 13)])
FIRST_MIXED = "SELECT id, created_at FROM events ORDER BY created_at DESC, id ASC LIMIT 4"


def walk_mixed(sql, args_for):
    seen, rows = [], db.execute(FIRST_MIXED).fetchall()
    while rows:
        seen += [row[0] for row in rows]
        rows = db.execute(sql, args_for(rows[-1])).fetchall()
    return seen


print()
for label, sql, args in (
        ("row value with DESC, ASC", ROW_VALUE, lambda r: (r[1], r[0])),
        ("expanded OR with DESC, ASC", EXPANDED, lambda r: (r[1], r[1], r[0]))):
    seen = walk_mixed(sql, args)
    print(f"{label}: {len(seen)} rows returned for 12 rows, repeated {sorted(set(i for i in seen if seen.count(i) > 1))}")

python step10_gotchas.py

due_at is NULL when unknown: 7 of 10 rows returned, never returned [8, 9, 10]
due_at is 0 when unknown: 10 of 10 rows returned, never returned []

row value with DESC, ASC: 17 rows returned for 12 rows, repeated [1, 3, 4, 6, 9, 10]
expanded OR with DESC, ASC: 12 rows returned for 12 rows, repeated []

Trap 1: a sort column that can be NULL. SQLite’s documentation says that “SQLite considers NULL values to be smaller than any other values for sorting purposes”, so in a descending sort the NULL rows come last, and then the cursor test stops working. In the row value rules, NULL means unknown, and “The overall result of comparison is NULL if it is possible to make the result either true or false by substituting alternative values in place of the constituent NULLs.” A comparison whose result is NULL does not select the row, so after the last row with a due date, the walk finds nothing more. The output shows 7 of 10 tickets returned and tickets 8, 9 and 10 never seen, with no error. The fix is to make the sort column NOT NULL and store a sentinel such as 0 for “unknown”; the second line shows all 10 returned.

Trap 2: sorting columns in different directions. A row value compares every column the same way. If you sort by created_at DESC, id ASC and use (created_at, id) < (?, ?), the tie part of the condition points the wrong way. The output shows 17 rows returned for a 12-row table, and six different rows were returned more than once. The correct form is the expanded condition, created_at < ? OR (created_at = ? AND id > ?), and it returned exactly 12. Remember what Step 6 showed: write that form with a redundant range in front of it (created_at <= ? for a descending first column) so the planner can seek, and check the plan: if it shows USE TEMP B-TREE FOR ORDER BY, SQLite is sorting the rows itself because the index does not match the mixed order.

Trap 3: what keyset pagination cannot do. The client cannot jump to page 40. Winand lists this as the main limitation, “most notably that you cannot directly navigate to arbitrary pages”, and it suits feeds, logs and infinite scrolling better than a numbered pager. It also gives you no cheap total. Google’s proposal allows a total but says “This total may be an estimate (but the API should explicitly document that).” Step 11 shows why an exact one is expensive.

Step 11: Measure the difference at every depth

Last, the table that decides whether the work was worth it. step11_bench.py reaches the same page by both methods at six depths. For each depth it looks up the cursor position that corresponds to the offset, checks that both queries return the same 100 rows, and times each with the median of nine runs.

"""step11_bench.py: the same page, reached by OFFSET and by a keyset cursor, at growing depth."""
import statistics
import time

from db import connect_ro
from pager import Position, fetch_page

OFFSET_SQL = ("SELECT id, created_at, kind, payload FROM events WHERE tenant_id = 1 "
              "ORDER BY created_at DESC, id DESC LIMIT 100 OFFSET ?")
POSITION_SQL = ("SELECT created_at, id FROM events WHERE tenant_id = 1 "
                "ORDER BY created_at DESC, id DESC LIMIT 1 OFFSET ?")


def median_ms(fn, repeats=9):
    fn()                                              # warm up
    times = []
    for _ in range(repeats):
        t0 = time.perf_counter()
        fn()
        times.append((time.perf_counter() - t0) * 1000)
    return statistics.median(times)


conn = connect_ro()
print(f"{'depth':>9}  {'OFFSET ms':>10}  {'keyset ms':>10}  same 100 rows")
for depth in (0, 1_000, 10_000, 100_000, 400_000, 800_000):
    pos = Position(*conn.execute(POSITION_SQL, (depth - 1,)).fetchone()) if depth else None
    offset_ids = [row[0] for row in conn.execute(OFFSET_SQL, (depth,))]
    keyset_ids = [row[0] for row in fetch_page(conn, 1, 100, pos)[0]]
    offset_ms = median_ms(lambda: conn.execute(OFFSET_SQL, (depth,)).fetchall())
    keyset_ms = median_ms(lambda: fetch_page(conn, 1, 100, pos))
    print(f"{depth:>9,}  {offset_ms:>10.2f}  {keyset_ms:>10.2f}  {offset_ids == keyset_ids}")

count_ms = median_ms(lambda: conn.execute("SELECT COUNT(*) FROM events WHERE tenant_id = 1").fetchone())
print(f"\nSELECT COUNT(*) FROM events WHERE tenant_id = 1 takes {count_ms:.2f} ms on its own")

python step11_bench.py

    depth   OFFSET ms   keyset ms  same 100 rows
        0        0.09        0.10  True
    1,000        0.11        0.10  True
   10,000        0.27        0.10  True
  100,000        1.81        0.10  True
  400,000       11.27        0.10  True
  800,000       22.83        0.10  True

SELECT COUNT(*) FROM events WHERE tenant_id = 1 takes 20.98 ms on its own

The OFFSET column grows with depth, from 0.09 to about 23 milliseconds. The keyset column stays at about 0.10 milliseconds at every depth, and the last column confirms that both returned the same rows each time. The final line is the cost of an exact total: counting tenant 1’s rows takes about 21 milliseconds on its own, more than two hundred times the cost of the page. That is why endpoints with big tables often skip the total or return an estimate. These numbers come from one Windows machine with SQLite running inside the Python process and a warm cache, so treat the absolute values as relative: a networked database adds latency to both columns, but the OFFSET column still grows with depth and the keyset column does not.

Step 12: Lock the behavior in with tests

Everything above was checked by running scripts. A test suite keeps it true when someone edits the code next month. Save test_pagination.py. It builds a 60-row database in a temporary folder with ten events per second, so ties straddle every page boundary. First the fixtures and two helpers:

"""test_pagination.py: the contract of the keyset endpoint."""
import base64
import hashlib
import hmac
import sqlite3

import pytest
from fastapi.testclient import TestClient

import app as app_module
from cursor import BadCursor, decode, encode, fingerprint
from db import INDEX_SQL, TABLE_SQL
from pager import Position, build_query

KEY = b"test-key"
H1, H2 = {"X-API-Key": "key-tenant-1"}, {"X-API-Key": "key-tenant-2"}


@pytest.fixture
def db(tmp_path, monkeypatch):
    """Tenant 1 owns ids 1-40, tenant 2 owns 41-60; ten events share each second."""
    conn = sqlite3.connect(tmp_path / "test.db")
    conn.executescript(TABLE_SQL + INDEX_SQL)
    conn.executemany(
        "INSERT INTO events VALUES (?, ?, ?, ?, ?)",
        [(i, 1 if i <= 40 else 2, 100 + i // 10, ["login", "upload"][i % 2], f"event-{i}")
         for i in range(1, 61)])
    conn.commit()
    monkeypatch.setenv("EVENTS_DB", str(tmp_path / "test.db"))
    yield conn
    conn.close()


@pytest.fixture
def client(db):
    return TestClient(app_module.app)


def get(client, headers=H1, **params):
    return client.get("/events", params=params, headers=headers)


def walk(client, headers=H1, first=None, **params):
    """Follow next_cursor to the end. `first` is a page the test already fetched."""
    ids, body = [], first or get(client, headers, **params).json()
    while True:
        ids += [item["id"] for item in body["items"]]
        if body["next_cursor"] is None:
            return ids
        body = get(client, headers, cursor=body["next_cursor"], **params).json()

The db fixture creates the table and the data and points the app at it through the EVENTS_DB environment variable. walk follows next_cursor to the end, the way a real client would. The tests about correct paging come first:

# test_pagination.py (continued)
def test_every_row_exactly_once_newest_first(client):
    assert walk(client, limit=7) == sorted(range(1, 41), reverse=True)


def test_ties_that_straddle_a_page_boundary_are_not_lost(client):
    for limit in (3, 4, 10, 11):
        assert sorted(walk(client, limit=limit)) == list(range(1, 41))


def test_rows_added_mid_walk_repeat_nothing(client, db):
    first = get(client, limit=7).json()
    db.executemany("INSERT INTO events VALUES (?, 1, ?, 'login', 'new')",
                   [(100 + k, 200 + k) for k in range(5)])
    db.commit()
    assert sorted(walk(client, first=first, limit=7)) == list(range(1, 41))


def test_deleting_the_row_a_cursor_points_at_is_harmless(client, db):
    first = get(client, limit=7).json()
    db.execute("DELETE FROM events WHERE id = ?", (first["items"][-1]["id"],))
    db.commit()
    assert sorted(walk(client, first=first, limit=7)) == list(range(1, 41))

These four tests are the heart of the suite: every row exactly once in the right order, ties across page boundaries (page sizes 3, 4, 10 and 11, chosen so the boundary lands in different places), rows inserted during a walk, and the deletion of the very row a cursor points at. Then the tests about limits, credentials and bad cursors:

# test_pagination.py (continued)
@pytest.mark.parametrize("limit", [0, -5, 101])
def test_limit_is_bounded(client, limit):
    assert get(client, limit=limit).status_code == 422


def test_missing_or_unknown_api_key_is_401(client):
    assert client.get("/events").status_code == 401
    assert client.get("/events", headers={"X-API-Key": "nope"}).status_code == 401


def test_a_tenant_only_sees_its_own_rows(client):
    assert set(walk(client, H2, limit=5)) == set(range(41, 61))


def test_tampered_cursor_is_rejected(client):
    body, tag = get(client, limit=5).json()["next_cursor"].split(".")
    forged = ("A" if body[0] != "A" else "B") + body[1:] + "." + tag
    assert get(client, cursor=forged, limit=5).status_code == 400


def test_cursor_is_bound_to_its_filter_and_tenant(client):
    cursor = get(client, limit=5, kind="login").json()["next_cursor"]
    assert get(client, cursor=cursor, limit=5, kind="upload").status_code == 400
    assert get(client, cursor=cursor, limit=5).status_code == 400
    assert get(client, H2, cursor=cursor, limit=5, kind="login").status_code == 400


@pytest.mark.parametrize("token", ["", "x", "a.b", "a.b.c", "....", "\xe9.\xe9"])
def test_garbage_is_400_not_500(client, token):
    assert get(client, cursor=token, limit=5).status_code == 400


def test_validly_signed_junk_is_rejected():
    body = b"[1, 2]"
    tag = hmac.new(KEY, body, hashlib.sha256).digest()
    token = (base64.urlsafe_b64encode(body).rstrip(b"=").decode() + "."
             + base64.urlsafe_b64encode(tag).rstrip(b"=").decode())
    with pytest.raises(BadCursor):
        decode(token, fingerprint(tenant=1, kind=None), KEY)


def test_round_trip_and_wrong_key():
    fp, pos = fingerprint(tenant=1, kind=None), Position(123, 456)
    assert decode(encode(pos, fp, KEY), fp, KEY) == pos
    with pytest.raises(BadCursor):
        decode(encode(pos, fp, KEY), fp, b"other-key")


def test_empty_result_and_exact_fit_have_no_cursor(client):
    assert get(client, kind="delete", limit=5).json() == {"items": [], "next_cursor": None}
    body = get(client, H2, limit=20).json()
    assert len(body["items"]) == 20 and body["next_cursor"] is None

The garbage-cursor test is parametrized over six strings, including an empty cursor and a non-ASCII one, and requires a 400 for each, not a 500: a malformed token is the client’s mistake and should never become a server crash. The signed-junk test calls decode directly with a validly signed body of the wrong shape, which no HTTP request could produce without the key. The last test is the plan gate from the EXPLAIN QUERY PLAN tutorial applied to this query: it fails if the page query ever stops using the index or starts needing a sort.

# test_pagination.py (continued)
def test_page_query_uses_the_index_and_needs_no_sort(db):
    for with_kind in (False, True):
        sql = build_query(kind=with_kind, after=True)
        args = [1] + (["login"] if with_kind else []) + [100, 5, 10]
        plan = " | ".join(row[3] for row in db.execute("EXPLAIN QUERY PLAN " + sql, args))
        assert "idx_events_tenant_time" in plan, plan
        assert "TEMP B-TREE" not in plan, plan

python -m pytest -q

..................... [100%]
21 passed in 0.98s

Twenty-one test cases pass, counting each parametrized value as its own case.

Step 13: Run it for real

The test client calls the app directly instead of opening a network port, so finish by starting the actual server and calling it with curl. In one terminal, set a key and start uvicorn, the server program that runs FastAPI apps:

set CURSOR_KEY=live-demo-key
python -m uvicorn app:app --port 8771

In PowerShell use $env:CURSOR_KEY = "live-demo-key", and on macOS and Linux use export CURSOR_KEY=live-demo-key. In a second terminal, repeat the requests below. step13_live.py does this automatically, starting the server, running the four requests, and stopping the server, and its output is what you should see:

"""step13_live.py: start the real server, page through it with curl, stop it."""
import json
import os
import subprocess
import sys
import time
import urllib.request

PORT = 8771
BASE = f"http://127.0.0.1:{PORT}/events"
env = dict(os.environ, CURSOR_KEY="live-demo-key")
server = subprocess.Popen([sys.executable, "-m", "uvicorn", "app:app", "--port", str(PORT),
                           "--log-level", "warning"], env=env)


def curl(url):
    return subprocess.run(["curl", "-s", "-H", "X-API-Key: key-tenant-2", url],
                          capture_output=True, text=True).stdout


try:
    for _ in range(100):
        try:
            urllib.request.urlopen(f"http://127.0.0.1:{PORT}/openapi.json", timeout=1)
            break
        except OSError:
            time.sleep(0.2)
    print(f"$ curl -s -H 'X-API-Key: key-tenant-2' '{BASE}?limit=2'")
    first = curl(f"{BASE}?limit=2")
    print(first)
    cursor = json.loads(first)["next_cursor"]
    print(f"\n$ curl -s -H 'X-API-Key: key-tenant-2' '{BASE}?limit=2&cursor=<next_cursor from above>'")
    print(curl(f"{BASE}?limit=2&cursor={cursor}"))
    print(f"\n$ curl -s -H 'X-API-Key: key-tenant-2' '{BASE}?limit=2&cursor=<next_cursor with its last 3 characters changed>'")
    print(curl(f"{BASE}?limit=2&cursor={cursor[:-3]}AAA"))
    print(f"\n$ curl -s -H 'X-API-Key: key-tenant-2' '{BASE}?limit=1000'")
    print(curl(f"{BASE}?limit=1000"))
finally:
    subprocess.run(["taskkill", "/PID", str(server.pid), "/T", "/F"], capture_output=True)

python step13_live.py

$ curl -s -H 'X-API-Key: key-tenant-2' 'http://127.0.0.1:8771/events?limit=2'
{"items":[{"id":999955,"created_at":1790124994,"kind":"delete","payload":"event-999955"},{"id":999948,"created_at":1790124993,"kind":"update","payload":"event-999948"}],"next_cursor":"eyJ2IjoxLCJjIjoxNzkwMTI0OTkzLCJpIjo5OTk5NDgsImYiOiI3Y2NkMjFiMTcyMTEwNzkwIn0.JAZgSoNHAUNa5MEC0M64MOEi9tWBWOL4V1Q2klPL3Cc"}

$ curl -s -H 'X-API-Key: key-tenant-2' 'http://127.0.0.1:8771/events?limit=2&cursor=<next_cursor from above>'
{"items":[{"id":999921,"created_at":1790124990,"kind":"logout","payload":"event-999921"},{"id":999844,"created_at":1790124980,"kind":"logout","payload":"event-999844"}],"next_cursor":"eyJ2IjoxLCJjIjoxNzkwMTI0OTgwLCJpIjo5OTk4NDQsImYiOiI3Y2NkMjFiMTcyMTEwNzkwIn0.RPj_GRLNE-2YimAN8nlPM2Axllzc9G0IuZuwefcmCxQ"}

$ curl -s -H 'X-API-Key: key-tenant-2' 'http://127.0.0.1:8771/events?limit=2&cursor=<next_cursor with its last 3 characters changed>'
{"detail":"invalid cursor"}

$ curl -s -H 'X-API-Key: key-tenant-2' 'http://127.0.0.1:8771/events?limit=1000'
{"detail":[{"type":"less_than_equal","loc":["query","limit"],"msg":"Input should be less than or equal to 100","input":"1000","ctx":{"le":100}}]}

The first request returns two events and a cursor. The second sends that cursor back and gets the next two events, which are older. The third changes the last three characters of the cursor and gets {"detail":"invalid cursor"}. The fourth asks for a thousand rows and gets the validation error. The same endpoint, called over HTTP by a program that has never seen your code, behaves the way the tests say it does.

Common mistakes and how to spot them

Confirm everything works end to end

With all the files saved, these commands run the whole tutorial from a clean folder. The comments say what each should show:

python step01_seed.py            # 1,000,000 rows, tenant 1 has 900,540
python step02_drift.py           # shown twice: [10, 9, 8] and never shown: [7, 6]
python step03_cost.py            # about 23 ms at offset 800,000
python step04_ties.py            # 31,840 rows never returned, then a stuck cursor
python step05_walk.py            # 0 repeated, 0 missing for keyset
python step06_plans.py           # row value about 0.1 ms, expanded OR about 33 ms
python step07_cursor.py          # six rejections
python step08_api.py             # 249 pages, 24,853 rows, six error responses
python step09_attack.py          # forged cursor returns tenant 1 rows, signed refuses
python step10_gotchas.py         # 7 of 10, 17 rows for 12
python step11_bench.py           # keyset flat at about 0.1 ms
python -m pytest -q              # 21 passed
python step13_live.py            # real server, real curl

If step02_drift.py or any step cannot import db or pager, you are running it from a folder other than the one that holds the files. If step08_api.py fails on the import of the test client, check that httpx2 is installed in the active environment. If the timings are different from the ones printed here, that is expected; what must match are the counts and the shape of the timing columns.

Where this stops working, and what to learn next

• Going backwards. The endpoint only moves toward older rows. A “previous page” cursor reverses the comparison and the sort, fetches the page, and flips the rows back. You can build it with the same tools, and the tests above are a good template for proving it.
• The sort key must not change. If you sort by a column that can be edited, such as updated_at, a row can move across the cursor between requests and be skipped or repeated. Use values that never change after creation, like the time of creation and the id, or accept the drift and document it.
• It is not a snapshot. New rows appear above the cursor and are not shown to a walk that has started. If you need a consistent export, record the highest id when you begin and ignore anything above it.
• Keys need a life cycle. This code has one signing key. To rotate it, add a key id to the token and keep the old key for verification until old cursors have expired. The same proposal describes expiring tokens after a reasonable time, which you could add with a timestamp in the signed body.
• Other databases. The idea works anywhere you can say “rows after this pair”, but the syntax and the planner’s behavior differ, so repeat Step 6 on your own database before trusting the speed.
• Related reading. The same signing and comparison idea protects incoming webhooks in the webhook signature tutorial. For conditional requests that save bandwidth on pages that have not changed, see the ETag and If-Match tutorial. If a page is expensive to build and popular, the cache stampede tutorial covers sharing one build among many callers.

You now have an endpoint that does not repeat or skip rows when the table changes underneath it, whose deepest page costs about the same as its first, whose cursors are rejected when they are edited or reused for another query, and a test suite and a plan check that will tell you the moment one of those properties breaks.

Sources and method

The behavior of SQLite row values, NULL comparison, ties and the cost of OFFSET comes from the SQLite row value and SELECT documentation, and the claim about skipped rows from the PostgreSQL LIMIT and OFFSET page. The drift problem and the keyset recipe are from Markus Winand’s keyset pagination post. The rules for opaque, non-authorizing page tokens and for keeping arguments the same are from Google’s AIP-158. The per-page limit is from OWASP’s API4:2023, the constant-time comparison from the Python hmac documentation, and the real-world cursor example from Stripe’s API reference. Every code block was executed, and the outputs shown are the saved outputs of those runs on Python 3.13.14, SQLite 3.50.4, FastAPI 0.143.0, Starlette 1.7.0, uvicorn 0.54.0 and pytest 9.1.1. Timings are medians of seven or nine runs on one Windows 11 machine with a warm cache. The lab data is synthetic and generated with a fixed seed, so the row counts repeat exactly while the millisecond values will not.