FastAPI 游标分页端点

示例说明

偏移分页在深页与并发写入下都不稳定，键集分页把锚点设为上一页末条记录的复合排序键：时间戳加 ID 双字段保证全序无歧义。游标令牌对客户端是不透明字符串——编码里附带的校验和让篡改与损坏在解码时立即暴露返回 400，服务端实现细节完全收敛在边界内。

技术要点

• 复合键 (at, id) 构造无歧义全序
• 游标带校验和防客户端篡改
• 多取一判定法免掉总数 count 查询

代码解读

encode_cursor 把时间戳与 ID 编码后追加哈希校验和再整体 base64；decode 校验失败抛 400。list_events 按游标过滤出锚点之前的记录，切出 limit 条；len 大于 limit 说明还有下一页，用本页末条生成下一游标，否则返回 None 表示终点。

使用建议

排序键要建复合索引支撑。游标应设置时效防长期重放。前端保留上页末游标即可实现无限滚动，无需页码状态。

代码示例

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,
    )