Designing a Compliance Score REST API

This walkthrough sits under Compliance Score APIs & Payload Contracts and solves one precise task: exposing the stored compliance results that the scoring pipeline already wrote — sliced by store, fixture, and date range — through a read API that a dashboard, a category-manager tool, or a partner integration can page through without ever double-counting or missing a row. The parent cluster owns the shape of the payload; this page owns the transport. The records themselves originate upstream in Planogram Sync & SKU Mapping Strategies, which emits the typed fields — compliance_percentage, out_of_stock_flags, misplaced_sku_list, registry_revision, capture_timestamp — that this API serves verbatim. A naive GET /scores?page=3 breaks the instant a new capture lands mid-scroll: offset paging shifts every row and clients silently skip or repeat records. The fix is a cursor-paginated, filterable, versioned, scope-guarded read endpoint built on FastAPI and Pydantic, and each part below is independently verifiable.

Prerequisites & Context Jump to heading#

Before building the endpoint, confirm the store below the API is already in place. This page serves records; it does not compute them.

• Runtime: Python 3.11+, FastAPI 0.110+, Pydantic v2, and an async database driver (asyncpg against Postgres in the examples). The compliance scores are already persisted — writing them is the concern of the upstream pipeline, not this read path.
• A stored, immutable score table: each row is one fixture evaluated at one capture_timestamp, keyed by a monotonically stable score id, and carrying the typed payload fields. Rows are append-only; a re-score writes a new row rather than mutating an old one, which is what lets cursors stay stable. Reset-driven schema shifts are handled by the sibling Versioning Compliance Payloads Across Store Resets.
• An index that matches the sort key: a composite index on (store_id, capture_timestamp, id) so the keyset predicate is a range scan, not a sort of the whole table.
• An auth provider: any OAuth2 / OIDC issuer that mints tokens carrying a scope claim. The API only verifies scopes; it does not issue them.

A note on terms: a cursor here is an opaque, forward-only pointer encoding the last row a client saw — not a page number. That single choice is what makes the endpoint correct under concurrent inserts.

Step 1 — Design the Resource and URL Jump to heading#

Model compliance results as a single flat collection resource, /scores, and express every slice as a query parameter rather than a nested path. A dashboard asks three questions — which store, which fixture, over what window — so store_id, fixture_id, from, and to are all filters on one collection, not separate endpoints. This keeps the surface small and the cache keys predictable.

GET /v1/scores?store_id=S-0148&fixture_id=F-22&from=2026-07-01T00:00:00Z&to=2026-07-08T00:00:00Z&limit=100

Rules that keep the resource clean:

• The collection is read-only. Only GET is defined; scores are produced upstream, so POST/PUT/DELETE never appear here.
• store_id is required; fixture_id is optional. A caller must scope to a store — an unbounded scan across every store is never a legitimate dashboard query — but may widen to all fixtures in that store.
• The window is half-open, [from, to). Inclusive-start, exclusive-end removes the double-count at midnight boundaries when a client walks day by day.
• limit is clamped. Accept a client limit but cap it server-side (default 100, max 500) so no single request can pull a store’s entire history.
• Version lives in the path prefix /v1, with header negotiation layered on in Step 4.

Step 2 — Type the Response With Pydantic Over the Payload Fields Jump to heading#

Serve the canonical payload fields exactly as the upstream pipeline emits them, and let Pydantic be the contract. A typed response model is the single source of truth for the OpenAPI schema, the validation, and the serialization, so drift between docs and reality is impossible.

from datetime import datetime
from pydantic import BaseModel, Field, ConfigDict


class ComplianceScore(BaseModel):
    """One fixture scored at one capture, served verbatim from storage."""
    model_config = ConfigDict(frozen=True)

    id: str
    store_id: str
    fixture_id: str
    compliance_percentage: float = Field(ge=0.0, le=100.0)
    out_of_stock_flags: list[str] = Field(default_factory=list)
    misplaced_sku_list: list[str] = Field(default_factory=list)
    registry_revision: int = Field(ge=1)
    capture_timestamp: datetime


class ScorePage(BaseModel):
    """A single page of results plus the cursor for the next one."""
    items: list[ComplianceScore]
    next_cursor: str | None = None
    limit: int

The frozen=True config makes each record immutable in memory, which matches its immutability in storage. registry_revision is served as-is so a client can tell which planogram revision a score was graded against — the field the sibling versioning page keys its reconciliation on. Because FastAPI derives the OpenAPI document from ScorePage, the published contract and the wire format cannot diverge.

Step 3 — Paginate With a Keyset Cursor and Apply Filters Jump to heading#

Replace OFFSET with a keyset (seek) cursor keyed on the exact sort tuple (capture_timestamp, id). The cursor encodes the last row of the page just served; the next query asks for rows strictly after that tuple. New inserts never shift a client’s position, because the pointer is a value, not a count. Encode it as opaque base64 so clients treat it as a token, not something to hand-edit.

import base64
import json
from datetime import datetime


def encode_cursor(last: "ComplianceScore") -> str:
    raw = json.dumps({"t": last.capture_timestamp.isoformat(), "id": last.id})
    return base64.urlsafe_b64encode(raw.encode()).decode()


def decode_cursor(token: str) -> tuple[datetime, str]:
    try:
        raw = json.loads(base64.urlsafe_b64decode(token.encode()))
        return datetime.fromisoformat(raw["t"]), raw["id"]
    except (ValueError, KeyError) as exc:
        raise ValueError("malformed cursor") from exc


async def fetch_page(
    conn, *, store_id: str, fixture_id: str | None,
    frm: datetime, to: datetime, limit: int, cursor: str | None,
) -> "ScorePage":
    """Keyset page over (capture_timestamp, id); stable under inserts."""
    clauses = ["store_id = $1", "capture_timestamp >= $2", "capture_timestamp < $3"]
    args: list[object] = [store_id, frm, to]

    if fixture_id is not None:
        args.append(fixture_id)
        clauses.append(f"fixture_id = ${len(args)}")

    if cursor is not None:
        c_ts, c_id = decode_cursor(cursor)
        args.extend([c_ts, c_id])
        # strict tuple comparison: rows after the last one seen
        clauses.append(f"(capture_timestamp, id) > (${len(args) - 1}, ${len(args)})")

    args.append(limit + 1)  # over-fetch one row to detect a next page
    sql = (
        f"SELECT * FROM compliance_scores WHERE {' AND '.join(clauses)} "
        f"ORDER BY capture_timestamp, id LIMIT ${len(args)}"
    )
    rows = await conn.fetch(sql, *args)

    items = [ComplianceScore(**dict(r)) for r in rows[:limit]]
    has_more = len(rows) > limit
    return ScorePage(
        items=items,
        next_cursor=encode_cursor(items[-1]) if has_more and items else None,
        limit=limit,
    )

Two details carry the correctness. The ORDER BY and the cursor comparison use the same tuple (capture_timestamp, id), so the seek predicate lines up with the index and never skips ties on a shared timestamp. And the query over-fetches exactly one row (limit + 1) to decide whether a next_cursor is warranted, rather than issuing a separate COUNT — a total count over an append-only stream is both expensive and instantly stale.

Step 4 — Negotiate the Version and Enforce Auth Scopes Jump to heading#

Two concerns bolt onto the handler: which contract version the client wants, and whether the token is allowed to read compliance data at all. Pin a major version in the path (/v1) for coarse routing, and allow an Accept header to request a minor revision within it — Accept: application/vnd.shelfanalytics.compliance+json; v=1 — so additive field changes ship without a breaking URL move. Reject an unsupported version explicitly rather than silently serving the default.

Auth is scope-gated: the token must carry compliance:read. A dependency validates the bearer token, extracts the scope claim, and rejects anything missing the scope with 403, before the handler ever touches the database.

from fastapi import FastAPI, Depends, Header, HTTPException, Query, Response
from datetime import datetime

app = FastAPI()
SUPPORTED_VERSIONS = {"1"}
REQUIRED_SCOPE = "compliance:read"


async def require_scope(authorization: str = Header(...)) -> dict:
    token = authorization.removeprefix("Bearer ").strip()
    claims = verify_token(token)              # raises on bad signature/expiry
    if REQUIRED_SCOPE not in claims.get("scope", "").split():
        raise HTTPException(403, f"missing scope {REQUIRED_SCOPE}")
    return claims


def negotiate_version(accept: str = Header("")) -> str:
    for part in accept.split(";"):
        part = part.strip()
        if part.startswith("v="):
            v = part[2:]
            if v not in SUPPORTED_VERSIONS:
                raise HTTPException(406, f"unsupported version {v}")
            return v
    return "1"  # default major when the client sends no version token


@app.get("/v1/scores", response_model=ScorePage)
async def list_scores(
    response: Response,
    store_id: str = Query(..., min_length=1),
    fixture_id: str | None = Query(None),
    frm: datetime = Query(..., alias="from"),
    to: datetime = Query(...),
    limit: int = Query(100, ge=1, le=500),
    cursor: str | None = Query(None),
    version: str = Depends(negotiate_version),
    claims: dict = Depends(require_scope),
) -> ScorePage:
    if frm >= to:
        raise HTTPException(422, "'from' must be strictly before 'to'")
    async with pool.acquire() as conn:
        page = await fetch_page(
            conn, store_id=store_id, fixture_id=fixture_id,
            frm=frm, to=to, limit=limit, cursor=cursor,
        )
    response.headers["ETag"] = weak_etag(page)   # over the serialized page
    response.headers["Cache-Control"] = "private, max-age=30"
    return page

The version and scope checks run as FastAPI dependencies, so an unauthenticated or wrongly-versioned request is rejected before any query executes — cheap failures stay cheap. The ETag is computed over the serialized page so a client re-polling an unchanged window can send If-None-Match and receive a 304 (wired in the middleware referenced below), which matters when a dashboard refreshes every 30 s.

Verification & Testing Jump to heading#

Assert each guarantee deterministically rather than trusting the happy path:

1. Pagination is stable under inserts. Fetch page one, insert a new score whose capture_timestamp sorts before the page-one boundary, then fetch page two with the returned cursor. Assert no id from page one reappears and none is skipped — the keyset pointer must ignore the late insert.
2. The window is half-open. Insert a score at exactly to; assert it is absent, and insert one at exactly from; assert it is present.
3. 304 on an unchanged repeat. Call the endpoint, capture the ETag, call again with If-None-Match set to it, and assert 304 Not Modified with an empty body. Then insert a matching score and assert the repeat returns 200 with a fresh ETag.
4. Filters are exact. Query one fixture_id and assert every returned row’s fixture_id matches; drop the filter and assert rows from sibling fixtures in the same store_id now appear.
5. Version negotiation. Send Accept: …; v=1 and assert 200; send v=9 and assert 406 Not Acceptable; send no version token and assert the default v=1 contract is served.
6. Scope enforcement. Present a token without compliance:read and assert 403 before any row is read (assert the query counter did not increment); present the correct scope and assert 200.

A healthy endpoint holds p95 < 250 ms on a 100-row page because every query is an index range scan with no OFFSET and no COUNT.

Troubleshooting Jump to heading#

Related Jump to heading#

• Compliance Score APIs & Payload Contracts — the parent cluster defining the payload shape this endpoint transports
• Versioning Compliance Payloads Across Store Resets — how registry_revision and schema changes are versioned behind the same API
• Planogram Sync & SKU Mapping Strategies — the upstream pillar that produces the compliance records this API serves