FastAPI Keyset Pagination Endpoint

import base64
import hashlib
from datetime import datetime

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel

app = FastAPI()

EVENTS = [
    {'id': i, 'kind': 'view' if i % 2 else 'click', 'at': f'2026-10-21T10:{i:02d}:00Z'}
    for i in range(50)
]


def encode_cursor(event: dict) -> str:
    raw = f"{event['at']}|{event['id']}".encode()
    checksum = hashlib.sha256(raw).hexdigest()[:8]
    return base64.urlsafe_b64encode(raw + b'|' + checksum.encode()).decode()


def decode_cursor(token: str) -> tuple[str, int]:
    try:
        decoded = base64.urlsafe_b64decode(token.encode()).decode()
        at, oid, checksum = decoded.rsplit('|', 2)
        raw = f'{at}|{oid}'.encode()
        if hashlib.sha256(raw).hexdigest()[:8] != checksum:
            raise ValueError('checksum mismatch')
        return at, int(oid)
    except Exception as exc:
        raise HTTPException(400, 'malformed cursor') from exc


class EventPage(BaseModel):
    items: list[dict]
    next_cursor: str | None


@app.get('/events', response_model=EventPage)
def list_events(
    limit: int = Query(20, ge=1, le=100),
    cursor: str | None = None,
):
    if cursor:
        at, oid = decode_cursor(cursor)
        filtered = [e for e in EVENTS if (e['at'], e['id']) < (at, oid)]
    else:
        filtered = EVENTS

    page = filtered[:limit]
    return EventPage(
        items=page,
        next_cursor=encode_cursor(page[-1]) if len(filtered) > limit else None,
    )