FastAPI cursor-baseret paginering — skalerbar API pagination

Offset-paginering bryder ned ved store datasæt og concurrency. Cursor-baseret paginering er standard for produktions-API'er med millioner af rækker.

---

Offset vs. cursor pagination

# OFFSET (undgå ved store tabeller):
SELECT * FROM readings ORDER BY timestamp DESC OFFSET 50000 LIMIT 50
# Problem: PostgreSQL scanner 50.050 rækker → O(N) cost
# Problem: Nye rækker tilføjet → side 2 overlapper med side 1

# CURSOR (keyset pagination):
SELECT * FROM readings
WHERE timestamp < '2025-01-15T06:00:00Z'   # Cursor position
ORDER BY timestamp DESC
LIMIT 50
# PostgreSQL bruger index direkte → O(log N) cost
# Stabil ved concurrent inserts

---

Opaque cursor implementation

# server/src/pagination.py

import base64
import json
from datetime import datetime
from typing import TypeVar, Generic
from pydantic import BaseModel

T = TypeVar("T")


class Page(BaseModel, Generic[T]):
    items: list[T]
    next_cursor: str | None = None
    has_more: bool


def encode_cursor(values: dict) -> str:
    """Konvertér sort-værdier til opaque cursor-string."""
    payload = json.dumps(values, default=str)
    return base64.urlsafe_b64encode(payload.encode()).decode()


def decode_cursor(cursor: str) -> dict:
    """Dekodér cursor tilbage til sort-værdier."""
    try:
        payload = base64.urlsafe_b64decode(cursor.encode()).decode()
        return json.loads(payload)
    except Exception:
        raise ValueError("Ugyldig cursor")


# Eksempel cursor ved aflæsninger sorteret på (timestamp DESC, id DESC):
# cursor = {"timestamp": "2025-01-15T06:00:00Z", "id": "550e8400-..."}

---

FastAPI endpoint med cursor pagination

# server/src/readings/router.py

from fastapi import APIRouter, Depends, Query
from sqlalchemy import select, and_
from server.src.pagination import Page, encode_cursor, decode_cursor
from server.src.db.models import Reading

router = APIRouter()


@router.get("/readings", response_model=Page[ReadingOut])
async def list_readings(
    cursor: str | None = Query(None, description="Pagination cursor"),
    limit: int = Query(50, ge=1, le=200),
    session: AsyncSession = Depends(get_session),
    current_user: User = Depends(require_role("landlord")),
) -> Page[ReadingOut]:
    stmt = (
        select(Reading)
        .where(Reading.tenant_id == current_user.tenant_id)
        .order_by(Reading.timestamp.desc(), Reading.id.desc())
        .limit(limit + 1)        # Hent én ekstra for has_more
    )

    if cursor:
        pos = decode_cursor(cursor)
        cursor_ts = datetime.fromisoformat(pos["timestamp"])
        cursor_id = pos["id"]
        # Keyset condition: (timestamp, id) < (cursor_ts, cursor_id)
        stmt = stmt.where(
            (Reading.timestamp < cursor_ts)
            | (
                (Reading.timestamp == cursor_ts)
                & (Reading.id < cursor_id)
            )
        )

    rows = (await session.execute(stmt)).scalars().all()
    has_more = len(rows) > limit
    items = rows[:limit]

    next_cursor = None
    if has_more and items:
        last = items[-1]
        next_cursor = encode_cursor({
            "timestamp": last.timestamp.isoformat(),
            "id": str(last.id),
        })

    return Page(
        items=[ReadingOut.model_validate(r) for r in items],
        next_cursor=next_cursor,
        has_more=has_more,
    )

---

Stable sort — undgå duplicate/missing rækker

# Regel: Sort key skal være unik + stabil

# FORKERT — timestamp alene er ikke unik:
.order_by(Reading.timestamp.desc())
# Problem: Samtidige reads med samme timestamp → ubestemt rækkefølge

# KORREKT — composite sort key (timestamp, id):
.order_by(Reading.timestamp.desc(), Reading.id.desc())
# UUID er altid unik → stabil cursor position

# For TimescaleDB hypertables:
# timestamp er chunk-key → index eksisterer automatisk
# Compound index (timestamp DESC, id DESC) anbefales:
CREATE INDEX readings_cursor_idx ON readings (timestamp DESC, id DESC);

---

Pydantic response schema

# server/src/readings/schemas.py

from pydantic import BaseModel, ConfigDict
from datetime import datetime
from decimal import Decimal
import uuid


class ReadingOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: uuid.UUID
    meter_installation_id: uuid.UUID
    timestamp: datetime
    value: Decimal
    unit: str
    rssi_dbm: int | None = None
    battery_level_pct: int | None = None


class ReadingsPage(BaseModel):
    items: list[ReadingOut]
    next_cursor: str | None = None
    has_more: bool
    total_hint: int | None = None    # Estimeret total (valgfri)

---

Frontend: TanStack Query infinite scroll

// ui/src/hooks/useReadings.ts

import { useInfiniteQuery } from "@tanstack/react-query";
import { api } from "../api";

export function useReadings(installationId: string) {
  return useInfiniteQuery({
    queryKey: ["readings", installationId],
    queryFn: ({ pageParam }) =>
      api.readings.list({
        installation_id: installationId,
        cursor: pageParam ?? undefined,
        limit: 50,
      }),
    initialPageParam: null as string | null,
    getNextPageParam: (lastPage) => lastPage.next_cursor ?? null,
    staleTime: 30_000,
  });
}

// Komponent:
function ReadingsList({ installationId }: { installationId: string }) {
  const { data, fetchNextPage, hasNextPage, isFetchingNextPage } =
    useReadings(installationId);

  const allReadings = data?.pages.flatMap((p) => p.items) ?? [];

  return (
    <div>
      {allReadings.map((r) => (
        <ReadingRow key={r.id} reading={r} />
      ))}
      {hasNextPage && (
        <button
          onClick={() => fetchNextPage()}
          disabled={isFetchingNextPage}
        >
          {isFetchingNextPage ? "Henter..." : "Vis flere"}
        </button>
      )}
    </div>
  );
}

---

Bidirektionel pagination (frem + tilbage)

# Til UI med "forrige side" funktion:

class BidirectionalPage(BaseModel, Generic[T]):
    items: list[T]
    next_cursor: str | None = None
    prev_cursor: str | None = None
    has_more: bool
    has_prev: bool


@router.get("/readings", response_model=BidirectionalPage[ReadingOut])
async def list_readings(
    cursor: str | None = None,
    direction: Literal["forward", "backward"] = "forward",
    limit: int = 50,
    ...
):
    if direction == "backward":
        # Reverser sort og keyset condition:
        stmt = stmt.order_by(Reading.timestamp.asc(), Reading.id.asc())
        # ... keyset condition med > i stedet for <
        items = list(reversed(items))    # Vend tilbage til korrekt rækkefølge
    ...

---

Konklusion

Cursor pagination med composite sort key (timestamp + id) er standard for IoT tidsserie-data. Opaque base64-encodede cursors skjuler implementeringsdetaljer og er sikre mod manipulation. TanStack useInfiniteQuery integrerer direkte med cursor-API'et. TimescaleDB's hypertable-indekser gør keyset queries effektive selv ved millioner af aflæsninger.

Se TanStack Query guide eller asyncpg guide.