python 51 lines · 9 steps

Cursor pagination in a FastAPI endpoint

An articles endpoint pages through rows with opaque, base64-encoded cursors instead of numeric offsets.

Explained by highlit

FastAPI Intermediate pagination cursor async base64 query-parameters

1import base64
2import json
3from typing import Annotated, Optional
4 
5from fastapi import APIRouter, Depends, HTTPException, Query
6from pydantic import BaseModel
7from sqlalchemy import select
8from sqlalchemy.ext.asyncio import AsyncSession
9 
10from .database import get_session
11from .models import Article
12from .schemas import ArticleOut
13 
14router = APIRouter(prefix="/articles", tags=["articles"])
15 
16 
17def encode_cursor(article_id: int) -> str:
18    raw = json.dumps({"id": article_id}).encode()
19    return base64.urlsafe_b64encode(raw).decode()
20 
21 
22def decode_cursor(cursor: str) -> int:
23    try:
24        payload = json.loads(base64.urlsafe_b64decode(cursor.encode()))
25        return int(payload["id"])
26    except (ValueError, KeyError, json.JSONDecodeError):
27        raise HTTPException(status_code=400, detail="Invalid pagination cursor")
28 
29 
30class ArticlePage(BaseModel):
31    items: list[ArticleOut]
32    next_cursor: Optional[str] = None
33 
34 
35@router.get("", response_model=ArticlePage)
36async def list_articles(
37    session: Annotated[AsyncSession, Depends(get_session)],
38    cursor: Optional[str] = None,
39    limit: int = Query(20, ge=1, le=100),
40):
41    stmt = select(Article).order_by(Article.id.asc()).limit(limit + 1)
42    if cursor is not None:
43        stmt = stmt.where(Article.id > decode_cursor(cursor))
44 
45    rows = (await session.scalars(stmt)).all()
46 
47    has_more = len(rows) > limit
48    page = rows[:limit]
49    next_cursor = encode_cursor(page[-1].id) if has_more else None
50 
51    return ArticlePage(items=page, next_cursor=next_cursor)

01 / 01

STEP 01

Walkthrough

Space play ←→ step click any line

Three takeaways

1Cursor pagination keeps queries efficient at scale because it filters by a keyed boundary instead of skipping rows with OFFSET.
2Fetching limit+1 rows is a cheap trick to detect whether another page exists without a separate count query.
3Encoding the cursor keeps its shape opaque to clients, so you can change what it contains without breaking their code.

Related explainers

python

from itertools import cycle
from collections import defaultdict

Round-robin task distribution in Python

round-robin iterators load-balancing

Intermediate 6 steps

typescript

import { useState, useEffect, useRef, useCallback } from "react";
 
interface Suggestion {
  id: string;

A debounced autocomplete hook in React

debounce custom-hooks abortcontroller

Advanced 7 steps

java

@RestController
@RequestMapping("/api/products")
@Validated
public class ProductSearchController {

Validating query params in a Spring controller

validation pagination rest-api

Intermediate 8 steps

python

import secrets
 
from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import APIKeyHeader

API key authentication as a FastAPI dependency

authentication dependency-injection api-keys

Intermediate 8 steps

python

import hashlib
from collections import defaultdict
from pathlib import Path

Finding duplicate files by size then hash

hashing file-io deduplication

Intermediate 7 steps

typescript

type Middleware<TIn, TOut> = (ctx: TIn) => Promise<TOut> | TOut;
 
class Pipeline<TIn, TOut> {
  private constructor(private readonly run: Middleware<TIn, TOut>) {}