import base64
import json
from datetime import datetime
from typing import Optional, Tuple
class InvalidCursor(Exception):
pass
def encode_cursor(created_at: datetime, item_id: int) -> str:
payload = json.dumps({"c": created_at.isoformat(), "i": item_id})
return base64.urlsafe_b64encode(payload.encode()).decode()
def decode_cursor(token: Optional[str]) -> Optional[Tuple[datetime, int]]:
if not token:
return None
try:
raw = base64.urlsafe_b64decode(token.encode()).decode()
data = json.loads(raw)
return datetime.fromisoformat(data["c"]), int(data["i"])
except (ValueError, KeyError, TypeError, json.JSONDecodeError) as exc:
raise InvalidCursor("malformed pagination cursor") from exc
from dataclasses import dataclass
from typing import List, Optional, Tuple
from sqlalchemy import tuple_
from sqlalchemy.orm import Query
from .cursor import encode_cursor
from .models import Order
@dataclass
class PageResult:
items: List[Order]
next_cursor: Optional[str]
def apply_cursor(query: Query, after: Optional[Tuple]) -> Query:
query = query.order_by(Order.created_at.asc(), Order.id.asc())
if after is not None:
last_created_at, last_id = after
query = query.filter(
tuple_(Order.created_at, Order.id) > (last_created_at, last_id)
)
return query
def keyset_page(query: Query, after: Optional[Tuple], limit: int) -> PageResult:
query = apply_cursor(query, after)
rows = query.limit(limit + 1).all()
has_more = len(rows) > limit
items = rows[:limit]
next_cursor = None
if has_more and items:
tail = items[-1]
next_cursor = encode_cursor(tail.created_at, tail.id)
return PageResult(items=items, next_cursor=next_cursor)
from datetime import datetime
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from sqlalchemy.orm import Session
from .cursor import InvalidCursor, decode_cursor
from .db import get_session
from .models import Order
from .paginate import keyset_page
router = APIRouter()
class OrderOut(BaseModel):
id: int
total_cents: int
created_at: datetime
class Config:
orm_mode = True
class Page(BaseModel):
items: List[OrderOut]
next_cursor: Optional[str]
@router.get("/orders", response_model=Page)
def list_orders(
cursor: Optional[str] = Query(default=None),
limit: int = Query(default=25, ge=1, le=100),
session: Session = Depends(get_session),
):
try:
after = decode_cursor(cursor)
except InvalidCursor:
raise HTTPException(status_code=400, detail="invalid cursor")
result = keyset_page(session.query(Order), after, limit)
return Page(items=result.items, next_cursor=result.next_cursor)
Cursor pagination (also called keyset pagination) avoids the classic OFFSET problem where deep pages get slower and rows shift under concurrent inserts. Instead of skipping N rows, it remembers the sort position of the last row seen and asks the database for rows strictly after it. This snippet wires up an opaque, tamper-resistant cursor for a FastAPI endpoint backed by SQLAlchemy.
The cursor.py tab defines the cursor contract. A cursor is just the ordering key of the last item in a page — here the tuple (created_at, id) — serialized to JSON and base64-encoded so the client treats it as an opaque token rather than a poke-able offset. encode_cursor and decode_cursor handle round-tripping, and decode_cursor raises InvalidCursor on any malformed input so the endpoint can turn it into a clean 400 rather than a 500. Encoding the full sort key (not just the id) is essential: sorting on a non-unique column like created_at requires the id tie-breaker to guarantee a total order, otherwise rows at a timestamp boundary can be skipped or duplicated.
The paginate.py tab holds the reusable query logic. apply_cursor builds the keyset predicate using a row-value comparison — (created_at, id) > (last_created_at, last_id) — expressed with SQLAlchemy's tuple_(...) so a single composite index on (created_at, id) satisfies it. The keyset_page helper always fetches limit + 1 rows: the extra row is a cheap probe that tells it whether a next page exists without a second COUNT query. If the extra row comes back, it is trimmed off and the last kept row becomes the next cursor.
The orders_api.py tab exposes GET /orders. It validates limit with Query bounds, decodes any incoming cursor, delegates to keyset_page, and returns a Page model containing the serialized items and a next_cursor that is None on the final page. A key trade-off worth noting: keyset pagination supports only forward (or reverse) traversal along a stable sort, not random jumps to page 42, and the sort columns must be immutable enough that a row's key does not move mid-scan. For infinite-scroll feeds and large tables, that constraint is well worth the consistent performance.
Related snips
class PostsController < ApplicationController
def index
@posts = Post.includes(:author)
.order(created_at: :desc)
.page(params[:page])
.per(10)Turbo Frames: infinite scroll with lazy-loading frame
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
Rails.application.configure do
config.after_initialize do
Bullet.enable = true
Bullet.alert = false
Bullet.bullet_logger = true
Bullet.console = trueN+1 query detection with Bullet gem
json.array! @posts do |post|
json.cache! ['v1', post], expires_in: 1.hour do
json.id post.id
json.title post.title
json.excerpt post.excerpt
json.published_at post.published_atFragment caching for expensive JSON serialization
class AddSettingsToAccounts < ActiveRecord::Migration[7.1]
disable_ddl_transaction!
def change
add_column :accounts, :settings, :jsonb, null: false, default: {}Postgres JSONB Partial Index for Feature Flags
Share this code
Here's the card — post it anywhere.
