import base64
import json
from datetime import datetime
from typing import NamedTuple
from fastapi import HTTPException
class Cursor(NamedTuple):
created_at: datetime
id: int
def encode_cursor(cursor: Cursor) -> str:
payload = json.dumps(
{"t": cursor.created_at.isoformat(), "i": cursor.id}
).encode("utf-8")
return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=")
def decode_cursor(raw: str) -> Cursor:
try:
padding = "=" * (-len(raw) % 4)
payload = base64.urlsafe_b64decode(raw + padding)
data = json.loads(payload)
return Cursor(
created_at=datetime.fromisoformat(data["t"]),
id=int(data["i"]),
)
except (ValueError, KeyError, TypeError):
raise HTTPException(status_code=400, detail="Invalid cursor")
from typing import Optional
from fastapi import Depends, Query
from .cursor import Cursor, decode_cursor
class CursorParams:
def __init__(
self,
limit: int = Query(20, ge=1, le=100),
cursor: Optional[str] = Query(None, description="Opaque next cursor"),
):
self.limit = limit
self._raw = cursor
@property
def after(self) -> Optional[Cursor]:
if self._raw is None:
return None
return decode_cursor(self._raw)
def pagination(params: CursorParams = Depends()) -> CursorParams:
return params
from fastapi import APIRouter, Depends, Request, Response
from sqlalchemy import select, tuple_
from sqlalchemy.ext.asyncio import AsyncSession
from .cursor import Cursor, encode_cursor
from .db import get_session
from .models import Article
from .pagination import CursorParams, pagination
from .schemas import ArticleOut
router = APIRouter()
@router.get("/articles", response_model=list[ArticleOut])
async def list_articles(
request: Request,
response: Response,
params: CursorParams = Depends(pagination),
db: AsyncSession = Depends(get_session),
):
stmt = select(Article).order_by(Article.created_at.desc(), Article.id.desc())
after = params.after
if after is not None:
stmt = stmt.where(
tuple_(Article.created_at, Article.id) < (after.created_at, after.id)
)
rows = (await db.execute(stmt.limit(params.limit + 1))).scalars().all()
has_next = len(rows) > params.limit
page = rows[: params.limit]
if has_next and page:
last = page[-1]
next_cursor = encode_cursor(Cursor(last.created_at, last.id))
_set_link_header(request, response, next_cursor, params.limit)
return page
def _set_link_header(request, response, cursor, limit):
url = request.url.replace_query_params(cursor=cursor, limit=limit)
response.headers["Link"] = f'<{url}>; rel="next"'
Cursor-based (keyset) pagination avoids the correctness and performance pitfalls of OFFSET/LIMIT paging: instead of skipping N rows, it anchors the next page on the last row's sort key, so inserts and deletes elsewhere in the table can't cause rows to be skipped or duplicated, and the query stays fast because the database can seek straight into the index. This snippet shows the whole flow in idiomatic FastAPI: a reusable cursor codec, a request-scoped dependency, and an endpoint that emits Link headers.
In cursor.py, a cursor is just an opaque, URL-safe base64 blob wrapping the tie-broken sort key — here the (created_at, id) pair. encode_cursor serializes that tuple to JSON and base64-encodes it so clients treat it as opaque and don't build their own offsets. decode_cursor reverses the process and raises HTTPException(400) on any malformed input, which keeps a tampered or truncated cursor from leaking a stack trace. Encoding an id alongside created_at is what makes the ordering total, so rows sharing a timestamp still page deterministically.
In pagination.py, CursorParams is a small dependency object built via Depends. It validates limit with Query bounds (1–100) and lazily decodes the incoming cursor only when accessed through the after property, so a request without a cursor pays no decoding cost. Modeling pagination as a dependency means every paginated route shares the same validation and default behavior for free.
In articles.py, list_articles fetches limit + 1 rows — the extra row is a cheap lookahead that reveals whether another page exists without a second COUNT query. The keyset predicate (created_at, id) < (:ts, :id) expresses the tie-broken comparison as a row-value comparison, which most databases optimize against a composite index. The handler trims the sentinel row, computes the next cursor from the last surviving row, and calls _set_link_header to write an RFC 5988 Link header with rel="next", mirroring how GitHub's API paginates. Clients follow the header rather than constructing URLs, so the server keeps full control over ordering and cursor format. The trade-off is that keyset paging can't jump to an arbitrary page number and needs a stable, indexed sort key — for infinite-scroll and feed-style APIs that's exactly the right shape.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
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
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
package com.example.myapp
import android.app.Application
import dagger.hilt.android.HiltAndroidApp
@HiltAndroidAppDependency injection with Hilt
Share this code
Here's the card — post it anywhere.
