Keyset Pagination Cursor
Paginate ordered rows by a keyset cursor instead of offset, keeping page cost stable as data grows.
Code
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)Report a problem with this exampleOpens a pre-filled issue on GitHub.