W4 Fixture — Sample API Payload + Warehouse DDL

Outcome: regenerate the pinned 200-row fixture locally and verify counts, uniqueness, and monotonicity — every later count reconciles against this.

There is no hosted fixture server. The fixture IS this generator: stdlib only, seed-pinned (SEED = 42), so every student builds against the same contract. Save as src/ingest/fixture.py and run it — no network.

Contract: 200 orders across 2 pages (per_page = 100) + 40 customers; order_id unique (o_000001–o_000200); updated_at monotonically non-decreasing across pages; status in {placed, shipped, delivered, refunded, cancelled}.

# src/ingest/fixture.py — deterministic W4 fixture (stdlib only, no network)
"""Seed-pinned orders + customers fixture. Run: python -m src.ingest.fixture --check"""
import json
import random
import sys
from datetime import datetime, timedelta, timezone

SEED = 42
PER_PAGE = 100
N_ORDERS = 200
N_CUSTOMERS = 40
STATUSES = ("placed", "shipped", "delivered", "refunded", "cancelled")
BASE_TS = datetime(2026, 9, 1, 10, 0, 0, tzinfo=timezone.utc)

def _iso(dt):
    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

def gen_customers(n=N_CUSTOMERS):
    rng = random.Random(SEED + 1)
    out, ts = [], BASE_TS
    for i in range(1, n + 1):
        ts = ts + timedelta(minutes=rng.randint(1, 30))
        out.append({"customer_id": f"c_{i:05d}",
                    "email": f"cust{i:05d}@example.in",
                    "updated_at": _iso(ts)})
    return out

_CUSTOMERS = None

def customers():
    global _CUSTOMERS
    if _CUSTOMERS is None:
        _CUSTOMERS = gen_customers()
    return _CUSTOMERS

def gen_orders(n=N_ORDERS):
    rng = random.Random(SEED)
    custs = customers()
    out, ts = [], BASE_TS
    for i in range(1, n + 1):
        ts = ts + timedelta(minutes=rng.randint(1, 12))
        c = custs[(i - 1) % len(custs)]
        out.append({"order_id": f"o_{i:06d}",
                    "customer_id": c["customer_id"],
                    "total": round(rng.uniform(99.0, 4999.0), 2),
                    "currency": "INR",
                    "status": STATUSES[rng.randrange(len(STATUSES))],
                    "updated_at": _iso(ts)})
    return out

_ORDERS = None

def orders():
    global _ORDERS
    if _ORDERS is None:
        _ORDERS = gen_orders()
    return _ORDERS

def get_page(resource, page, per_page=PER_PAGE):
    rows = orders() if resource == "orders" else customers()
    start = (page - 1) * per_page
    return {"page": page, "per_page": per_page, "total": len(rows),
            "data": rows[start:start + per_page]}

def check():
    errs = []
    os_ = orders()
    if len(os_) != N_ORDERS:
        errs.append(f"orders={len(os_)}, want {N_ORDERS}")
    ids = [o["order_id"] for o in os_]
    if len(set(ids)) != len(ids):
        errs.append("order_id not unique")
    ts = [o["updated_at"] for o in os_]
    if any(b < a for a, b in zip(ts, ts[1:])):
        errs.append("updated_at not monotonic")
    bad = [o["status"] for o in os_ if o["status"] not in STATUSES]
    if bad:
        errs.append(f"bad statuses: {sorted(set(bad))}")
    p1, p2, p3 = (get_page("orders", n)["data"] for n in (1, 2, 3))
    if not (len(p1) == 100 and len(p2) == 100 and p3 == []):
        errs.append("pagination != [100, 100, empty]")
    if errs:
        print("FIXTURE FAIL:", *errs, sep="\n  - ")
        return 1
    print(f"FIXTURE OK: {len(os_)} orders / 2 pages, "
          f"{len(customers())} customers, unique ids, monotonic updated_at")
    return 0

if __name__ == "__main__":
    if "--check" in sys.argv:
        sys.exit(check())
    print(json.dumps(get_page("orders", 1), indent=1)[:800] + "\n...")

Warehouse staging DDL (fixture)

CREATE TABLE IF NOT EXISTS stg_orders (
  order_id TEXT PRIMARY KEY,
  customer_id TEXT NOT NULL,
  total NUMERIC NOT NULL,
  currency CHAR(3) NOT NULL DEFAULT 'INR',
  status TEXT NOT NULL,
  updated_at TIMESTAMPTZ NOT NULL,
  loaded_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS stg_customers (
  customer_id TEXT PRIMARY KEY,
  email TEXT NOT NULL,
  updated_at TIMESTAMPTZ NOT NULL,
  loaded_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS ingest_watermark (
  source TEXT PRIMARY KEY,
  last_seen TIMESTAMPTZ NOT NULL
);

Expected outcome on the fixture

• First run: python -m src.ingest.fixture --check prints FIXTURE OK — 200 stg_orders rows after first load (2 pages x 100).
• Second run with no new data: row counts unchanged (idempotent).
• A mutated row (status change, bumped updated_at) upserts exactly one row.