功能说明
该示例实现键集分页：游标编码上一页最后一行的排序键，下一页从该键之后继续取。相比偏移量分页，它不随页号增大而扫描并跳过大量行，因此深页依然快，且不会因并发插入导致行重复或遗漏。游标是自包含的编码串，无需服务端保存会话状态。
关键代码
排序键取“创建时间 + 主键”组合，保证顺序确定且唯一，避免同一时间戳下的不确定排列。查询用元组比较 (created_at, id) > (?, ?) 推进，多取一行以判断是否还有下一页，随行数与游标一同返回。游标用 base64 包装 JSON，便于在网络传输且不可直接读懂。解码失败抛明确错误而非静默返回首页。
使用注意
游标是不透明的，客户端不应自行解析或修改，否则可能构造出任意跳转或注入内容；服务端应校验并限制其含义。排序键必须有唯一性兜底，只用创建时间会在时间戳重复时漏行或重复。键集分页不支持随机跳页，若产品需要页码导航则需另设方案。对可删除数据，删除行会导致边界变化，游标仍安全但顺序可能跨越被删区间。
代码示例
from dataclasses import dataclass
import base64
import json
from typing import Optional
@dataclass(frozen=True)
class Cursor:
created_at: str
row_id: int
def encode(cursor: Optional[Cursor]) -> Optional[str]:
if cursor is None:
return None
raw = json.dumps({'c': cursor.created_at, 'i': cursor.row_id}).encode('utf-8')
return base64.urlsafe_b64encode(raw).decode('ascii')
def decode(token: Optional[str]) -> Optional[Cursor]:
if not token:
return None
try:
data = json.loads(base64.urlsafe_b64decode(token.encode('ascii')))
return Cursor(created_at=str(data['c']), row_id=int(data['i']))
except (ValueError, KeyError, TypeError):
raise ValueError('invalid pagination cursor')
def fetch_page(conn, cursor: Optional[Cursor], limit: int = 20):
if cursor is None:
rows = conn.execute(
'SELECT id, created_at FROM items ORDER BY created_at, id LIMIT ?', (limit + 1,)).fetchall()
else:
rows = conn.execute(
'SELECT id, created_at FROM items WHERE (created_at, id) > (?, ?) '
'ORDER BY created_at, id LIMIT ?', (cursor.created_at, cursor.row_id, limit + 1)).fetchall()
has_more = len(rows) > limit
page = rows[:limit]
next_cursor = Cursor(page[-1][1], page[-1][0]) if has_more and page else None
return page, encode(next_cursor)