import base64
import json
from datetime import datetime
class InvalidCursor(Exception):
pass
def encode_cursor(created_at, row_id):
payload = {"t": created_at.isoformat(), "id": row_id}
raw = json.dumps(payload, separators=(",", ":")).encode("utf-8")
return base64.urlsafe_b64encode(raw).decode("ascii")
def decode_cursor(token):
try:
raw = base64.urlsafe_b64decode(token.encode("ascii"))
payload = json.loads(raw)
created_at = datetime.fromisoformat(payload["t"])
row_id = int(payload["id"])
except (ValueError, KeyError, TypeError):
raise InvalidCursor("cursor is malformed")
return created_at, row_id
from sqlalchemy import tuple_
from .models import Article, db
MAX_LIMIT = 100
def paginate_articles(limit, cursor=None):
limit = max(1, min(limit, MAX_LIMIT))
query = db.session.query(Article).order_by(
Article.created_at.desc(), Article.id.desc()
)
if cursor is not None:
created_at, row_id = cursor
# Row-value comparison keeps the seek correct across equal timestamps.
query = query.filter(
tuple_(Article.created_at, Article.id)
< tuple_(created_at, row_id)
)
# Fetch one extra row to know whether another page exists.
rows = query.limit(limit + 1).all()
has_more = len(rows) > limit
page = rows[:limit]
return page, has_more
from flask import Blueprint, jsonify, request
from .cursor import InvalidCursor, decode_cursor, encode_cursor
from .queries import paginate_articles
articles_bp = Blueprint("articles", __name__, url_prefix="/api/articles")
@articles_bp.errorhandler(InvalidCursor)
def handle_invalid_cursor(exc):
return jsonify(error="invalid_cursor", message=str(exc)), 400
@articles_bp.route("", methods=["GET"])
def list_articles():
limit = request.args.get("limit", default=20, type=int)
cursor = None
token = request.args.get("cursor")
if token:
cursor = decode_cursor(token)
page, has_more = paginate_articles(limit, cursor)
next_cursor = None
if has_more and page:
last = page[-1]
next_cursor = encode_cursor(last.created_at, last.id)
return jsonify(
data=[a.to_dict() for a in page],
page_info={"has_more": has_more, "next_cursor": next_cursor},
)
This snippet shows how to build stable, efficient pagination for a JSON API using keyset (cursor) pagination instead of LIMIT/OFFSET. Offset pagination degrades on large tables because the database still scans and discards every skipped row, and rows shifting between requests cause items to be duplicated or missed. Keyset pagination avoids both problems by remembering the last row seen and asking for rows strictly after it, using an indexed ordering column.
In cursor.py, the cursor is a small opaque token rather than a raw offset. encode_cursor serializes the ordering values — a timestamp and the row id as a tiebreaker — into URL-safe base64 JSON, and decode_cursor reverses it while treating any malformed input as a 400 via InvalidCursor. Encoding the values keeps the API contract loose: clients pass the token back verbatim without depending on its internal shape, which lets the server evolve the sort key later.
In queries.py, paginate_articles implements the actual keyset seek. Because ordering is on (created_at, id) descending, the WHERE clause uses a row-value comparison expressed with tuple_(...) < tuple_(...), which correctly handles rows sharing the same created_at. This compound key is essential — ordering by a non-unique column alone would let the cursor land ambiguously between equal rows. It fetches limit + 1 rows so it can detect whether a further page exists without a separate COUNT, then trims the extra row and reports has_more.
In articles_bp.py, the Flask blueprint wires this into an endpoint. list_articles clamps the client-supplied limit to a sane maximum to prevent unbounded queries, decodes an optional cursor, and calls the query helper. The response embeds a next_cursor built from the last returned row, so clients follow the chain by echoing that value. The InvalidCursor handler returns a clean JSON 400 rather than leaking a stack trace.
The trade-off is that keyset pagination only supports forward/backward stepping, not random jumps to page N, and requires an index on the sort tuple. For feeds, timelines, and infinite scroll — where users move sequentially — it is the more scalable and correct choice.
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
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
module EmailNormalization
extend ActiveSupport::Concern
included do
attr_accessor :soft_warningsSoft Validation: Normalize + Validate Email
class CreateTopSellersMv < ActiveRecord::Migration[7.0]
def up
execute <<~SQL
CREATE MATERIALIZED VIEW top_sellers AS
SELECT p.id AS product_id,
p.name AS product_name,Cache-Friendly “Top N” with Materialized View Refresh
Share this code
Here's the card — post it anywhere.
