Hands-on lab — Day 135: From API to DataFrame

"""A small paginated JSON API, built only from the standard library.

Everything in this lab talks to THIS server, on the loopback address
127.0.0.1, on a port the operating system picks at run time. Nothing here
opens a connection to the internet, and nothing here needs one.

The server mimics the shape of a real order-history API: customers arrive
one page at a time, each customer carries a list of orders (a one-to-many
nesting), and one field is missing from the early pages and only appears
from page 3 onward -- the schema-drift case this lab detects on purpose.

Endpoints:

  GET /api/customers?page=N&page_size=K
      Page N (1-indexed) of the full customer list, K per page. Answers
      {"page", "page_size", "total_pages", "customers": [...]}.

  GET /api/customers/incremental?since=ISO8601
      Every customer whose updated_at is greater than or equal to `since`
      (inclusive lower bound -- see the lesson for why), sorted by
      updated_at ascending. Answers {"customers": [...], "watermark": ISO}
      where watermark is the updated_at of the last record returned, or
      `since` unchanged if nothing matched.

  GET /control/stats
      {"requests": N} -- how many requests this server has answered since
      the last reset. This is what proves a "replay from raw" step made
      zero additional calls.

  GET /control/reset
      Zeroes the request counter without touching the dataset.

Run it on its own if you want to poke at it by hand:

    python3 examples/api_server.py

It prints the address it bound to and serves until you press Ctrl-C.
"""

from __future__ import annotations

import contextlib
import json
import socket
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Iterator
from urllib.parse import parse_qs, urlparse

# The seven-customer dataset. Page 1 and 2 (customers C1-C4) carry no
# `loyalty_tier` field at all -- not null, ABSENT -- and it first appears on
# C5, the first customer on page 3. That absence is what a schema-drift
# detector has to notice: pandas will backfill the missing key as NaN once
# the frame is assembled, and the six earlier rows will look fine unless
# something is watching for exactly this.
CUSTOMERS: list[dict[str, object]] = [
    {
        "customer_id": "C1",
        "name": "Ada Lovelace",
        "updated_at": "2026-01-05T10:00:00Z",
        "total_amount_due": "500.00",
        "orders": [
            {"order_id": "O1", "amount": "200.00", "status": "paid"},
            {"order_id": "O2", "amount": "300.00", "status": "paid"},
        ],
    },
    {
        "customer_id": "C2",
        "name": "Grace Hopper",
        "updated_at": "2026-01-06T10:00:00Z",
        "total_amount_due": "750.00",
        "orders": [{"order_id": "O3", "amount": "750.00", "status": "paid"}],
    },
    {
        "customer_id": "C3",
        "name": "Alan Turing",
        "updated_at": "2026-01-07T10:00:00Z",
        "total_amount_due": "300.00",
        "orders": [
            {"order_id": "O4", "amount": "100.00", "status": "paid"},
            {"order_id": "O5", "amount": "100.00", "status": "refunded"},
            {"order_id": "O6", "amount": "100.00", "status": "paid"},
        ],
    },
    {
        "customer_id": "C4",
        "name": "Katherine Johnson",
        "updated_at": "2026-01-08T10:00:00Z",
        "total_amount_due": "0.00",
        "orders": [],
    },
    {
        "customer_id": "C5",
        "name": "Margaret Hamilton",
        "updated_at": "2026-01-09T10:00:00Z",
        "total_amount_due": "420.00",
        "loyalty_tier": "gold",
        "orders": [{"order_id": "O7", "amount": "420.00", "status": "paid"}],
    },
    {
        "customer_id": "C6",
        "name": "Radia Perlman",
        "updated_at": "2026-01-10T10:00:00Z",
        "total_amount_due": "150.00",
        "loyalty_tier": "silver",
        "orders": [{"order_id": "O8", "amount": "150.00", "status": "paid"}],
    },
    {
        "customer_id": "C7",
        "name": "Hedy Lamarr",
        "updated_at": "2026-01-11T10:00:00Z",
        "total_amount_due": "610.00",
        "loyalty_tier": "gold",
        "orders": [{"order_id": "O9", "amount": "610.00", "status": "paid"}],
    },
]

PAGE_SIZE_DEFAULT = 2


class CountingServer(ThreadingHTTPServer):
    """A threading server that counts every request it answers.

    This counter is what turns "the replay touched no network" from a claim
    into something a test can read back and assert on.
    """

    daemon_threads = True
    allow_reuse_address = True

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self.lock = threading.Lock()
        self.requests = 0


class CustomerAPIHandler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    server_version = "DayLab/1.0"
    sys_version = ""

    def log_message(self, fmt: str, *args) -> None:  # noqa: D102 - silence stderr
        pass

    def version_string(self) -> str:
        return self.server_version

    def _send_json(self, status: int, payload: object) -> None:
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:  # noqa: N802 - name fixed by BaseHTTPRequestHandler
        parsed = urlparse(self.path)
        query = parse_qs(parsed.query)
        with self.server.lock:
            self.server.requests += 1

        if parsed.path == "/api/customers":
            self._page(query)
        elif parsed.path == "/api/customers/incremental":
            self._incremental(query)
        elif parsed.path == "/control/stats":
            with self.server.lock:
                self._send_json(200, {"requests": self.server.requests})
        elif parsed.path == "/control/reset":
            with self.server.lock:
                self.server.requests = 0
            self._send_json(200, {"requests": 0})
        else:
            self._send_json(404, {"error": "not_found", "path": parsed.path})

    def _page(self, query: dict[str, list[str]]) -> None:
        page = int(query.get("page", ["1"])[0])
        page_size = int(query.get("page_size", [str(PAGE_SIZE_DEFAULT)])[0])
        total_pages = -(-len(CUSTOMERS) // page_size)  # ceiling division
        start = (page - 1) * page_size
        rows = CUSTOMERS[start : start + page_size]
        self._send_json(
            200,
            {"page": page, "page_size": page_size, "total_pages": total_pages, "customers": rows},
        )

    def _incremental(self, query: dict[str, list[str]]) -> None:
        since = query.get("since", ["1970-01-01T00:00:00Z"])[0]
        matched = [c for c in CUSTOMERS if str(c["updated_at"]) >= since]
        matched.sort(key=lambda c: str(c["updated_at"]))
        watermark = str(matched[-1]["updated_at"]) if matched else since
        self._send_json(200, {"customers": matched, "watermark": watermark})


def wait_until_accepting(host: str, port: int, timeout: float = 5.0) -> None:
    deadline = time.monotonic() + timeout
    last_error: OSError | None = None
    while time.monotonic() < deadline:
        try:
            with socket.create_connection((host, port), timeout=0.25):
                return
        except OSError as exc:
            last_error = exc
            time.sleep(0.01)
    raise RuntimeError(f"the local test server never became ready on port {port}: {last_error}")


@contextlib.contextmanager
def running_server() -> Iterator[CountingServer]:
    """Start the server on an ephemeral port and shut it down afterwards."""
    server = CountingServer(("127.0.0.1", 0), CustomerAPIHandler)
    thread = threading.Thread(target=server.serve_forever, name="api-server", daemon=True)
    thread.start()
    try:
        wait_until_accepting(*server.server_address[:2])
        yield server
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)


def base_url(server: CountingServer) -> str:
    host, port = server.server_address[:2]
    return f"http://{host}:{port}"


if __name__ == "__main__":
    with running_server() as srv:
        print(f"serving on {base_url(srv)} -- press Ctrl-C to stop")
        try:
            while True:
                time.sleep(0.5)
        except KeyboardInterrupt:
            print("\nstopped.")

"""From API to DataFrame -- the nine ideas this lab tests, as real code.

Every function here works on plain Python objects (dicts and lists decoded
from JSON) and pandas DataFrames. Nothing here is specific to the mock
server in `api_server.py`; the fetch helpers just happen to talk to it with
`urllib.request`, which is the one stdlib HTTP client this lesson exercises
end to end.

Read this top to bottom in the order the lesson's nine exercises use it:

  1-2. flatten_customer_grain / flatten_order_grain -- the grain trap
  3.   explode_list_column                          -- nested lists
  4.   pin_dtypes                                    -- untyped arrival
  5.   detect_schema_drift / assemble_pages          -- drift across pages
  6.   fetch_raw_pages / transform_from_raw           -- raw before transform
  7.   upsert                                         -- idempotent ingestion
  8.   ContractViolation / check_contract              -- the boundary contract
  9.   fetch_incremental                               -- the watermark
"""

from __future__ import annotations

import json
import urllib.request
from pathlib import Path
from typing import Any

import pandas as pd

# --------------------------------------------------------------------------
# 1-2. The grain trap: two flattenings of the same nested payload.
# --------------------------------------------------------------------------


def flatten_customer_grain(customers: list[dict[str, Any]]) -> pd.DataFrame:
    """One row per customer. `orders` stays a Python list inside each cell.

    This is what you get from `pandas.json_normalize(customers)` with no
    `record_path`: it flattens each customer's own fields into columns but
    leaves any nested list exactly where it was. Every customer-level number
    -- `total_amount_due` here -- appears exactly once, which is what makes
    this the correct grain for a "how much is each customer's balance"
    question.
    """
    return pd.json_normalize(customers)


def flatten_order_grain(customers: list[dict[str, Any]]) -> pd.DataFrame:
    """One row per order. Every customer-level field is repeated by design.

    `record_path="orders"` tells `json_normalize` which nested list becomes
    the grain of the output; `meta` names the parent fields to carry down
    onto every child row. That carrying-down is not a bug -- it is exactly
    what `meta` is for -- but it means `total_amount_due` is now duplicated
    once per order, and `.sum()` on it no longer means what it meant a
    moment ago. A customer with no orders contributes zero rows here: an
    empty list under `record_path` produces zero rows for that customer
    (contrast this with `DataFrame.explode`, exercise 3, which keeps a row).
    """
    return pd.json_normalize(
        customers,
        record_path="orders",
        meta=["customer_id", "name", "total_amount_due"],
    )


def duplicated_meta_columns(customers: list[dict[str, Any]]) -> dict[str, int]:
    """For each `meta` column, how many times does each customer's value repeat?

    Returns {customer_id: number of order rows that customer contributed},
    which is exactly the duplication factor for every meta column on that
    customer's rows.
    """
    return {str(c["customer_id"]): len(c.get("orders", [])) for c in customers}


# --------------------------------------------------------------------------
# 3. Nested lists and explode.
# --------------------------------------------------------------------------


def explode_list_column(df: pd.DataFrame, column: str) -> pd.DataFrame:
    """Turn one row holding a list into one row per list element.

    Unlike `json_normalize(..., record_path=...)`, `DataFrame.explode`
    KEEPS a row for an empty list -- it produces a single row with NaN in
    the exploded column, rather than dropping the parent entirely. That
    difference is the point of running both this and the grain-trap
    functions above side by side.
    """
    return df.explode(column, ignore_index=True)


# --------------------------------------------------------------------------
# 4. Untyped arrival: everything from JSON is a string, a number, or None.
# --------------------------------------------------------------------------


def pin_dtypes(df: pd.DataFrame) -> tuple[pd.DataFrame, int]:
    """Coerce the known numeric and datetime columns; report how many values moved.

    Returns the pinned frame and the count of cells that were successfully
    converted from a string to a number across every column pinned here.
    Day 121's dtype-pinning discipline, applied to a frame that arrived as
    JSON rather than CSV -- the wrinkle here is that a column absent from
    some records (like `loyalty_tier`) is silently all-NaN for those rows
    rather than raising, so pinning has to tolerate missing columns too.
    """
    out = df.copy()
    coerced = 0

    for column in ("total_amount_due", "amount"):
        if column not in out.columns:
            continue
        before_numeric = pd.to_numeric(out[column], errors="coerce")
        was_string = out[column].apply(lambda v: isinstance(v, str))
        now_numeric = before_numeric.notna()
        coerced += int((was_string & now_numeric).sum())
        out[column] = before_numeric

    if "updated_at" in out.columns:
        out["updated_at"] = pd.to_datetime(out["updated_at"], utc=True, format="ISO8601")

    return out, coerced


# --------------------------------------------------------------------------
# 5. Schema drift across pages.
# --------------------------------------------------------------------------


def detect_schema_drift(pages: list[list[dict[str, Any]]]) -> dict[str, int]:
    """For each field that is NOT present on every page's first record set,
    report the 1-indexed page it first appears on.

    A field is "drift" here if it is missing from at least one earlier page
    and present on a later one. Fields present everywhere, or absent
    everywhere, are not drift.
    """
    seen_from: dict[str, int] = {}
    seen_by_page: list[set[str]] = []
    for page_records in pages:
        fields: set[str] = set()
        for record in page_records:
            fields |= set(record.keys())
        seen_by_page.append(fields)

    all_fields: set[str] = set()
    for fields in seen_by_page:
        all_fields |= fields

    drift: dict[str, int] = {}
    for field in sorted(all_fields):
        first_page = next(
            (i + 1 for i, fields in enumerate(seen_by_page) if field in fields), None
        )
        absent_from_an_earlier_page = any(field not in fields for fields in seen_by_page)
        if first_page is not None and absent_from_an_earlier_page and first_page > 1:
            drift[field] = first_page
        seen_from[field] = first_page or 0

    return drift


def assemble_pages(pages: list[list[dict[str, Any]]]) -> pd.DataFrame:
    """Concatenate every page's customer-grain flattening into one frame.

    A field that appears only from page 3 onward becomes a column that is
    NaN for every row from pages 1 and 2 -- pandas does this silently, which
    is exactly why `detect_schema_drift` exists as a separate, deliberate
    check rather than relying on someone noticing the NaNs.
    """
    frames = [flatten_customer_grain(page) for page in pages if page]
    if not frames:
        return pd.DataFrame()
    return pd.concat(frames, ignore_index=True)


# --------------------------------------------------------------------------
# 6. Raw-then-transform.
# --------------------------------------------------------------------------


def fetch_raw_pages(base_url: str, page_size: int, raw_path: Path) -> int:
    """Fetch every page from the API and persist each raw response as one
    JSONL line, before any transformation happens. Returns the number of
    HTTP requests made.
    """
    requests_made = 0
    page = 1
    total_pages = 1
    with raw_path.open("w", encoding="utf-8") as fh:
        while page <= total_pages:
            url = f"{base_url}/api/customers?page={page}&page_size={page_size}"
            with urllib.request.urlopen(url, timeout=5) as response:
                payload = json.loads(response.read().decode("utf-8"))
            requests_made += 1
            total_pages = payload["total_pages"]
            fh.write(json.dumps(payload) + "\n")
            page += 1
    return requests_made


def transform_from_raw(raw_path: Path) -> pd.DataFrame:
    """Rebuild the assembled, dtype-pinned frame from the stored raw JSONL,
    touching no network at all.
    """
    pages: list[list[dict[str, Any]]] = []
    with raw_path.open("r", encoding="utf-8") as fh:
        for line in fh:
            payload = json.loads(line)
            pages.append(payload["customers"])
    df = assemble_pages(pages)
    pinned, _ = pin_dtypes(df)
    return pinned


# --------------------------------------------------------------------------
# 7. Idempotent ingestion.
# --------------------------------------------------------------------------


def upsert(existing: pd.DataFrame, incoming: pd.DataFrame, key: str) -> pd.DataFrame:
    """Merge `incoming` into `existing`, keyed on `key`.

    Running this twice with the same `incoming` must leave the row count
    and the frame unchanged -- that is what "idempotent" means here. The
    incoming rows win a conflict, since they represent the most recently
    fetched state of that key.
    """
    if existing.empty:
        merged = incoming.copy()
    else:
        stays = existing[~existing[key].isin(incoming[key])]
        merged = pd.concat([stays, incoming], ignore_index=True)
    return merged.sort_values(key, ignore_index=True)


# --------------------------------------------------------------------------
# 8. The contract on the assembled frame.
# --------------------------------------------------------------------------


class ContractViolation(ValueError):
    """Raised by `check_contract` with the name of the rule that failed."""


REQUIRED_COLUMNS = {"customer_id", "name", "updated_at", "total_amount_due"}
MIN_ROWS, MAX_ROWS = 1, 10_000


def check_contract(df: pd.DataFrame) -> None:
    """Raise `ContractViolation` naming the first rule the frame breaks.

    Checked in order: required columns present, `customer_id` unique,
    `total_amount_due` numeric and non-negative, row count within bounds.
    """
    missing = REQUIRED_COLUMNS - set(df.columns)
    if missing:
        raise ContractViolation(f"missing required columns: {sorted(missing)}")

    if df["customer_id"].duplicated().any():
        dupes = sorted(df.loc[df["customer_id"].duplicated(), "customer_id"].unique())
        raise ContractViolation(f"duplicate customer_id: {dupes}")

    if not pd.api.types.is_numeric_dtype(df["total_amount_due"]):
        raise ContractViolation("total_amount_due is not numeric -- pin_dtypes must run first")

    if (df["total_amount_due"] < 0).any():
        raise ContractViolation("total_amount_due contains a negative balance")

    if not (MIN_ROWS <= len(df) <= MAX_ROWS):
        raise ContractViolation(f"row count {len(df)} is outside [{MIN_ROWS}, {MAX_ROWS}]")


# --------------------------------------------------------------------------
# 9. Incremental fetch by watermark.
# --------------------------------------------------------------------------


def fetch_incremental(base_url: str, since: str) -> tuple[pd.DataFrame, str]:
    """Fetch every customer updated at or after `since`.

    The boundary convention is INCLUSIVE (`>=`, not `>`): the server may
    hand back the same boundary record twice across two consecutive calls,
    but `upsert`'s natural-key merge absorbs that duplicate for free. The
    exclusive convention (`>`) would avoid the duplicate but risks silently
    DROPPING a record that shares its `updated_at` with the watermark record
    -- a customer updated in the same second the last page was fetched simply
    never appears in any later call. A harmless duplicate beats permanent
    data loss, so this lab errs on the side of `>=`.
    """
    url = f"{base_url}/api/customers/incremental?since={since}"
    with urllib.request.urlopen(url, timeout=5) as response:
        payload = json.loads(response.read().decode("utf-8"))
    df = flatten_customer_grain(payload["customers"]) if payload["customers"] else pd.DataFrame(
        columns=["customer_id", "name", "updated_at", "total_amount_due"]
    )
    return df, payload["watermark"]

"""The nine exercises, asserted against real behaviour.

Run with: pytest examples -q  (from the lab directory)

Every test in this file names the number of the exercise it belongs to in
its docstring, matching the lesson's "One Row Means One Thing" lab brief.
"""

from __future__ import annotations

import json
from pathlib import Path

import pandas as pd
import pytest

from api_server import CUSTOMERS, base_url, running_server
from ingest import (
    ContractViolation,
    assemble_pages,
    check_contract,
    detect_schema_drift,
    duplicated_meta_columns,
    explode_list_column,
    fetch_incremental,
    fetch_raw_pages,
    flatten_customer_grain,
    flatten_order_grain,
    pin_dtypes,
    transform_from_raw,
    upsert,
)

# A small standalone nested payload for exercises 1-3, independent of the
# server's dataset -- these exercises only need recorded JSON, not HTTP.
GRAIN_PAYLOAD = [
    {
        "customer_id": "C1",
        "name": "Ada Lovelace",
        "total_amount_due": 500.00,
        "orders": [
            {"order_id": "O1", "amount": "200.00"},
            {"order_id": "O2", "amount": "300.00"},
        ],
    },
    {
        "customer_id": "C2",
        "name": "Grace Hopper",
        "total_amount_due": 750.00,
        "orders": [{"order_id": "O3", "amount": "750.00"}],
    },
    {
        "customer_id": "C3",
        "name": "Alan Turing",
        "total_amount_due": 300.00,
        "orders": [
            {"order_id": "O4", "amount": "100.00"},
            {"order_id": "O5", "amount": "100.00"},
            {"order_id": "O6", "amount": "100.00"},
        ],
    },
]

TAGGED_PAYLOAD = pd.DataFrame(
    {
        "customer_id": ["C1", "C2", "C3"],
        "tags": [["vip", "early-adopter"], ["vip"], []],
    }
)


# --------------------------------------------------------------------------
# 1. The grain trap.
# --------------------------------------------------------------------------


def test_exercise1_the_two_flattenings_give_different_row_counts():
    customer_grain = flatten_customer_grain(GRAIN_PAYLOAD)
    order_grain = flatten_order_grain(GRAIN_PAYLOAD)

    assert len(customer_grain) == 3
    assert len(order_grain) == 6

    true_total = customer_grain["total_amount_due"].sum()
    inflated_total = order_grain["total_amount_due"].sum()

    assert true_total == 1550.0
    assert inflated_total == 2650.0
    assert inflated_total - true_total == 1100.0


# --------------------------------------------------------------------------
# 2. json_normalize with meta -- the duplication, understood not discovered.
# --------------------------------------------------------------------------


def test_exercise2_meta_columns_duplicate_by_the_order_count():
    order_grain = flatten_order_grain(GRAIN_PAYLOAD)
    duplication = duplicated_meta_columns(GRAIN_PAYLOAD)

    assert duplication == {"C1": 2, "C2": 1, "C3": 3}

    for customer_id, expected_repeats in duplication.items():
        rows = order_grain[order_grain["customer_id"] == customer_id]
        assert len(rows) == expected_repeats
        assert rows["total_amount_due"].nunique() == 1  # one value, repeated

    # Every meta column carries the duplication, not just total_amount_due.
    for column in ("customer_id", "name", "total_amount_due"):
        assert column in order_grain.columns


# --------------------------------------------------------------------------
# 3. explode: exact multiplication, empty list survives as NaN.
# --------------------------------------------------------------------------


def test_exercise3_explode_multiplies_rows_by_list_length():
    exploded = explode_list_column(TAGGED_PAYLOAD, "tags")

    # 2 tags + 1 tag + 1 (empty list keeps one row) = 4.
    assert len(exploded) == 4

    c3_rows = exploded[exploded["customer_id"] == "C3"]
    assert len(c3_rows) == 1
    assert pd.isna(c3_rows["tags"].iloc[0])

    c1_rows = exploded[exploded["customer_id"] == "C1"]
    assert len(c1_rows) == 2
    assert set(c1_rows["tags"]) == {"vip", "early-adopter"}


def test_exercise3_record_path_drops_what_explode_keeps():
    # Contrast: json_normalize(record_path=...) on the same shape of data
    # DROPS a record with an empty list entirely, rather than keeping a row.
    no_orders = [{"customer_id": "C4", "orders": []}]
    order_grain = pd.json_normalize(no_orders, record_path="orders", meta=["customer_id"])
    assert len(order_grain) == 0

    as_frame = pd.DataFrame(no_orders)
    exploded = explode_list_column(as_frame, "orders")
    assert len(exploded) == 1
    assert pd.isna(exploded["orders"].iloc[0])


# --------------------------------------------------------------------------
# 4. Untyped arrival.
# --------------------------------------------------------------------------


def test_exercise4_numeric_fields_arrive_as_strings_and_pinning_fixes_them():
    order_grain = flatten_order_grain(GRAIN_PAYLOAD)
    assert order_grain["amount"].apply(lambda v: isinstance(v, str)).all()

    pinned, coerced = pin_dtypes(order_grain)
    assert pd.api.types.is_numeric_dtype(pinned["amount"])
    assert coerced == 6  # every one of the 6 order-grain amount values
    assert pinned["amount"].sum() == pytest.approx(1550.0)


# --------------------------------------------------------------------------
# 5. Schema drift across pages.
# --------------------------------------------------------------------------


def _pages_from_server(page_size: int = 2) -> list[list[dict]]:
    with running_server() as server:
        url = base_url(server)
        pages: list[list[dict]] = []
        import urllib.request

        page = 1
        total_pages = 1
        while page <= total_pages:
            with urllib.request.urlopen(
                f"{url}/api/customers?page={page}&page_size={page_size}", timeout=5
            ) as response:
                payload = json.loads(response.read().decode("utf-8"))
            pages.append(payload["customers"])
            total_pages = payload["total_pages"]
            page += 1
        return pages


def test_exercise5_drift_detector_names_the_field_and_first_page():
    pages = _pages_from_server(page_size=2)
    assert len(pages) == 4  # 7 customers, page_size 2 -> pages of 2,2,2,1

    drift = detect_schema_drift(pages)
    assert drift == {"loyalty_tier": 3}

    assembled = assemble_pages(pages)
    assert len(assembled) == 7
    assert assembled["loyalty_tier"].isna().sum() == 4  # C1-C4 have none
    assert assembled["loyalty_tier"].notna().sum() == 3  # C5-C7 have it


# --------------------------------------------------------------------------
# 6. Raw then transform.
# --------------------------------------------------------------------------


def test_exercise6_raw_is_written_before_transform_and_replay_touches_no_server(tmp_path: Path):
    raw_path = tmp_path / "raw_customers.jsonl"

    with running_server() as server:
        url = base_url(server)
        requests_made = fetch_raw_pages(url, page_size=2, raw_path=raw_path)

    assert requests_made == 4  # one request per page, 4 pages
    assert raw_path.exists()
    lines = raw_path.read_text(encoding="utf-8").splitlines()
    assert len(lines) == 4  # one raw line per page, written before any transform

    # Replaying from the stored raw copy must not touch the server at all --
    # there is no server running here, so any attempted call would raise.
    replayed = transform_from_raw(raw_path)
    assert len(replayed) == 7
    assert pd.api.types.is_numeric_dtype(replayed["total_amount_due"])


# --------------------------------------------------------------------------
# 7. Idempotent ingestion.
# --------------------------------------------------------------------------


def test_exercise7_ingesting_the_same_page_twice_does_not_duplicate():
    page = CUSTOMERS[0:2]
    frame = flatten_customer_grain(page)
    pinned, _ = pin_dtypes(frame)

    once = upsert(pd.DataFrame(), pinned, key="customer_id")
    twice = upsert(once, pinned, key="customer_id")

    assert len(once) == 2
    assert len(twice) == 2
    pd.testing.assert_frame_equal(
        once.reset_index(drop=True), twice.reset_index(drop=True)
    )


def test_exercise7_upsert_replaces_a_changed_row_rather_than_adding_one():
    original = flatten_customer_grain([CUSTOMERS[0]])
    changed = flatten_customer_grain([CUSTOMERS[0]]).copy()
    changed.loc[0, "total_amount_due"] = "999.00"

    merged = upsert(original, changed, key="customer_id")
    assert len(merged) == 1
    assert merged.loc[0, "total_amount_due"] == "999.00"


# --------------------------------------------------------------------------
# 8. The contract.
# --------------------------------------------------------------------------


def test_exercise8_a_healthy_frame_passes_the_contract():
    pages = _pages_from_server(page_size=3)
    assembled = assemble_pages(pages)
    pinned, _ = pin_dtypes(assembled)
    check_contract(pinned)  # must not raise


def test_exercise8_a_corrupted_payload_is_named_and_refused():
    pages = _pages_from_server(page_size=3)
    assembled = assemble_pages(pages)
    pinned, _ = pin_dtypes(assembled)

    corrupted = pd.concat([pinned, pinned.iloc[[0]]], ignore_index=True)  # duplicate key
    with pytest.raises(ContractViolation, match="duplicate customer_id"):
        check_contract(corrupted)

    missing_column = pinned.drop(columns=["total_amount_due"])
    with pytest.raises(ContractViolation, match="missing required columns"):
        check_contract(missing_column)

    negative_balance = pinned.copy()
    negative_balance.loc[0, "total_amount_due"] = -5.0
    with pytest.raises(ContractViolation, match="negative balance"):
        check_contract(negative_balance)


# --------------------------------------------------------------------------
# 9. Incremental watermark.
# --------------------------------------------------------------------------


def test_exercise9_incremental_fetch_returns_only_records_after_the_watermark():
    with running_server() as server:
        url = base_url(server)
        first_batch, watermark = fetch_incremental(url, since="1970-01-01T00:00:00Z")
        assert len(first_batch) == 7
        assert watermark == "2026-01-11T10:00:00Z"

        second_batch, watermark2 = fetch_incremental(url, since=watermark)
        # Inclusive boundary: the watermark record (C7) comes back again.
        assert len(second_batch) == 1
        assert set(second_batch["customer_id"]) == {"C7"}
        assert watermark2 == watermark

        after_c7, _ = fetch_incremental(url, since="2026-01-11T10:00:01Z")
        assert len(after_c7) == 0

lesson_id: D135
day: 135
kind: python-program
languages: [python, bash]
setup_commands:
  - cd labs/sections/math-statistics-and-data/day-135-from-api-to-dataframe
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
  - .venv/bin/pytest --version
  - '.venv/bin/python3 -c "import pandas; print(pandas.__version__)"'
run_commands:
  - '.venv/bin/python3 examples/api_server.py   # serves on 127.0.0.1:<ephemeral> until Ctrl-C'
  - .venv/bin/pytest examples -q
  - .venv/bin/pytest starter -q
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - rm -rf .venv
  - 'git checkout -- starter/  # optional: reset your work'
requires_network: true
requires_api_key: false
estimated_minutes: 30
last_executed: '2026-08-20'
executed_on: 'macOS, Python 3.14.0, pandas 3.0.5, pytest 9.1.1, bash 3.2.57 -> bash tests/run_tests.sh -> 39 checks, 0 failure(s), exit 0. requires_network is true for the one-time dependency install ONLY; the tests and examples open no socket to anything but 127.0.0.1 on an operating-system-assigned port.'

# Dependencies -- Day 135 lab

**One third-party package. The API you talk to is not one of them -- it is
built entirely from the standard library.**

## The pinned list

`requirements.txt` contains exactly two lines:

```
pandas==3.0.5
pytest==9.1.1
```

| Dependency | Version | Licence | Why this lab needs it |
| --- | --- | --- | --- |
| pandas | 3.0.5 | BSD 3-Clause | `pandas.json_normalize`, `DataFrame.explode`, `pandas.to_numeric`, `pandas.to_datetime`, `pandas.concat` -- everything this lesson's ingestion pipeline flattens, types and assembles with. |
| pytest | 9.1.1 | MIT | The test runner. This lab needs fixtures, `pytest.raises`, `tmp_path`, and skip-on-unfinished markers. |

Both are free and open source. Neither has a paid tier, an account, or
telemetry. Both versions were installed and verified on the authoring
machine on 2026-08-20.

## What is NOT in the list, and why that matters

**The API.** `examples/api_server.py` is a subclass of
`http.server.BaseHTTPRequestHandler` served by `ThreadingHTTPServer`, with
`socket`, `threading` and `json` doing the rest -- standard library. You
already have it.

**The HTTP client.** This lab fetches every page with
`urllib.request.urlopen`, also standard library. No `requests`, no `httpx`.
The lesson names both as leading options and says plainly that neither ran
here: `urllib.request` is the one this lab actually exercises, because a
day about flattening JSON does not need a third-party HTTP client to prove
the flattening.

**pydantic.** The lesson describes schema validation with pydantic from its
documentation. It is not installed in the authoring virtual environment and
no pydantic code was run for this lab; every assertion of shape in this lab
is the `check_contract` function in `examples/ingest.py`, written by hand
against pandas dtypes.

## Install once

From this lab's directory:

```bash
python3 -m venv .venv
.venv/bin/pip install -r requirements/requirements.txt
.venv/bin/pytest --version
.venv/bin/python3 -c "import pandas; print(pandas.__version__)"
```

Those last two commands should print `pytest 9.1.1` and `3.0.5`.

## The network, precisely

The install needs the network **once**, to download two packages from the
Python Package Index. After that, nothing in this lab touches the internet.
Every socket opened by the tests or the examples goes to `127.0.0.1`, the
loopback interface, on a port the operating system assigned at run time --
never a hard-coded port, and never a real host.

If you already have both packages somewhere else, skip the virtual
environment and point the suite at that pytest:

```bash
PYTEST=/path/to/pytest bash tests/run_tests.sh
```

The runner resolves `python3` from the same directory as the pytest it
found, so the examples import the same pandas the tests do.

## Check your Python

```bash
python3 --version
```

Verified on 3.14.0. Python 3.10 or newer is required, because the code uses
the `X | None` annotation style.

pandas==3.0.5
pytest==9.1.1

# The brief -- One Row Means One Thing

You have been handed access to a small customer-and-orders API
(`examples/api_server.py`, started for you by `conftest.py`). It returns
paginated, nested JSON: each page is a list of customers, and each customer
carries a list of their own orders.

Your job is to build the ingestion pipeline in `ingest.py`, one function at
a time, that turns that nested JSON into a DataFrame you can trust: the
right grain, the right dtypes, a detector for the field that only shows up
from page 3 onward, a raw-before-transform step, an idempotent upsert, a
contract on the assembled frame, and an incremental fetch by watermark.

## Why the grain comes first

Before you write a single line, answer this question out loud: **does one
row in your target frame mean one customer, or one order?**

Both are legitimate. Neither is more "correct" than the other in general.
But they give different row counts and different aggregates from the
*same* JSON, and the difference is not a rounding error -- it is a customer
balance duplicated once per order that customer has. Get the grain wrong
and every downstream number is wrong in a way that looks completely
plausible until someone checks it against the invoice.

## The nine exercises

1. **`flatten_customer_grain` and `flatten_order_grain`** -- the two
   flattenings of the same nested payload, and the numbers that prove they
   disagree.
2. Understand exactly which columns `meta` duplicates, and by how much.
3. **`explode_list_column`** -- and see, on your own machine, what
   pandas 3.0.5 does with an empty list (it is not what `record_path`
   does with one).
4. **`pin_dtypes`** -- everything from JSON arrives as a string, a number,
   or `None`. Fix it, and count what you fixed.
5. **`detect_schema_drift`** -- one field is missing from the first six
   customers and present on the last three. Name it and the page it first
   appeared on.
6. Use the provided `fetch_raw_pages` / `transform_from_raw` to see raw
   storage pay for itself: a full re-run from disk touches no network.
7. **`upsert`** -- ingest the same page twice and prove nothing duplicated.
8. **`check_contract`** -- a frame that passes, and a frame that does not,
   with the exact rule named in the exception.
9. Use the provided `fetch_incremental` and explain, in your own words, why
   it errs toward a possible duplicate rather than a possible loss.

## Run your work at any time

```bash
.venv/bin/pytest starter -q
```

Unfinished exercises are skipped, so this is green from the first minute.
Delete one `raise NotImplementedError` at a time and watch a skip turn into
a pass.

"""Provided for you, complete. Do not edit -- this is the harness, not the work.

Puts the lab's `examples/` directory on the import path (so you can import
`api_server`), and starts ONE local HTTP server for the whole test session
on an ephemeral port on 127.0.0.1.

Read `examples/api_server.py` once before you start. It is the API you are
writing an ingestion pipeline against, and knowing its dataset -- seven
customers, `loyalty_tier` first appearing on the third page -- makes every
exercise below easier.
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Iterator

import pytest

LAB_DIR = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(LAB_DIR / "examples"))
sys.path.insert(0, str(Path(__file__).resolve().parent))

from api_server import CountingServer, base_url, running_server  # noqa: E402


@pytest.fixture(scope="session")
def server() -> Iterator[CountingServer]:
    with running_server() as srv:
        yield srv


@pytest.fixture(scope="session")
def base(server: CountingServer) -> str:
    return base_url(server)

"""YOUR FILE -- exercises 1, 2, 3, 4, 5, 7 and 8.

Seven functions below have a docstring saying exactly what they must do, a
signature that is already right, and a `raise NotImplementedError` you
delete. `examples/ingest.py` contains a complete reference implementation:
use it when you are stuck, but write yours first.

`fetch_raw_pages`, `transform_from_raw` and `fetch_incremental` (exercises 6
and 9) are provided complete below -- they are mostly HTTP and file
plumbing, not the pandas ideas this lab is about, and you will use them
as-is while writing the rest.

Run your work at any time:

    .venv/bin/pytest starter -q

Unfinished exercises are skipped, so the suite exits 0 from the first
minute. Check everything at the end with:

    bash tests/run_tests.sh
"""

from __future__ import annotations

import json
import urllib.request
from pathlib import Path
from typing import Any

import pandas as pd


def flatten_customer_grain(customers: list[dict[str, Any]]) -> pd.DataFrame:
    """Exercise 1a. One row per customer, using `pandas.json_normalize`.

    Call `pandas.json_normalize` with NO `record_path` -- that is what keeps
    each customer's nested `orders` list inside a single cell instead of
    exploding it, which is what makes this the customer grain.
    """
    raise NotImplementedError


def flatten_order_grain(customers: list[dict[str, Any]]) -> pd.DataFrame:
    """Exercise 1b. One row per order, using `record_path` and `meta`.

    Call `pandas.json_normalize` with `record_path="orders"` and
    `meta=["customer_id", "name", "total_amount_due"]`. Every meta field
    will repeat once per order that customer has.
    """
    raise NotImplementedError


def explode_list_column(df: pd.DataFrame, column: str) -> pd.DataFrame:
    """Exercise 3. One row per list element, using `DataFrame.explode`.

    Use `df.explode(column, ignore_index=True)`. Unlike `record_path`
    above, exploding an empty list KEEPS one row with NaN rather than
    dropping the parent record.
    """
    raise NotImplementedError


def pin_dtypes(df: pd.DataFrame) -> tuple[pd.DataFrame, int]:
    """Exercise 4. Coerce `total_amount_due` and `amount` to numeric, and
    `updated_at` to a UTC datetime, wherever those columns are present.

    Return `(pinned_df, coerced_count)` where `coerced_count` is the number
    of string cells across those numeric columns that successfully became
    numbers. Use `pandas.to_numeric(..., errors="coerce")` and
    `pandas.to_datetime(..., utc=True, format="ISO8601")`.
    """
    raise NotImplementedError


def detect_schema_drift(pages: list[list[dict[str, Any]]]) -> dict[str, int]:
    """Exercise 5. For each field missing from at least one EARLIER page but
    present on a LATER one, report the 1-indexed page it first appears on.

    A field present on every page, or on none, is not drift and must not
    appear in the returned dict.
    """
    raise NotImplementedError


def upsert(existing: pd.DataFrame, incoming: pd.DataFrame, key: str) -> pd.DataFrame:
    """Exercise 7. Merge `incoming` into `existing`, keyed on `key`.

    Rows in `existing` whose key also appears in `incoming` must be
    replaced, not duplicated. Running this twice with the same `incoming`
    must produce the same result both times -- that is the idempotence this
    exercise proves.
    """
    raise NotImplementedError


class ContractViolation(ValueError):
    """Raised by `check_contract` with the name of the rule that failed."""


REQUIRED_COLUMNS = {"customer_id", "name", "updated_at", "total_amount_due"}
MIN_ROWS, MAX_ROWS = 1, 10_000


def check_contract(df: pd.DataFrame) -> None:
    """Exercise 8. Raise `ContractViolation` naming the FIRST rule broken.

    Check, in this order: (1) every column in `REQUIRED_COLUMNS` is
    present, (2) `customer_id` has no duplicates, (3) `total_amount_due` is
    numeric, (4) `total_amount_due` has no negative values, (5) the row
    count is within `[MIN_ROWS, MAX_ROWS]`. Put the offending value(s) in
    the message, e.g. "duplicate customer_id: [...]".
    """
    raise NotImplementedError


# --------------------------------------------------------------------------
# Provided, complete -- exercises 6 and 9 build on these directly.
# --------------------------------------------------------------------------


def assemble_pages(pages: list[list[dict[str, Any]]]) -> pd.DataFrame:
    frames = [flatten_customer_grain(page) for page in pages if page]
    if not frames:
        return pd.DataFrame()
    return pd.concat(frames, ignore_index=True)


def fetch_raw_pages(base_url: str, page_size: int, raw_path: Path) -> int:
    """Exercise 6. Fetch every page and persist each raw response as one
    JSONL line BEFORE any transformation. Returns the number of requests made.
    Provided complete -- read it, then use it.
    """
    requests_made = 0
    page = 1
    total_pages = 1
    with raw_path.open("w", encoding="utf-8") as fh:
        while page <= total_pages:
            url = f"{base_url}/api/customers?page={page}&page_size={page_size}"
            with urllib.request.urlopen(url, timeout=5) as response:
                payload = json.loads(response.read().decode("utf-8"))
            requests_made += 1
            total_pages = payload["total_pages"]
            fh.write(json.dumps(payload) + "\n")
            page += 1
    return requests_made


def transform_from_raw(raw_path: Path) -> pd.DataFrame:
    """Exercise 6. Rebuild the frame from stored raw JSONL, touching no
    network. Provided complete -- it calls your `assemble_pages` and
    `pin_dtypes` once those are written.
    """
    pages: list[list[dict[str, Any]]] = []
    with raw_path.open("r", encoding="utf-8") as fh:
        for line in fh:
            payload = json.loads(line)
            pages.append(payload["customers"])
    df = assemble_pages(pages)
    pinned, _ = pin_dtypes(df)
    return pinned


def fetch_incremental(base_url: str, since: str) -> tuple[pd.DataFrame, str]:
    """Exercise 9. Fetch every customer updated at or after `since`
    (inclusive boundary -- see the lesson for why). Provided complete.
    """
    url = f"{base_url}/api/customers/incremental?since={since}"
    with urllib.request.urlopen(url, timeout=5) as response:
        payload = json.loads(response.read().decode("utf-8"))
    if payload["customers"]:
        df = flatten_customer_grain(payload["customers"])
    else:
        df = pd.DataFrame(columns=["customer_id", "name", "updated_at", "total_amount_due"])
    return df, payload["watermark"]

"""YOUR FILE -- checks that grade exercises 1 to 9.

Run it at any time:

    .venv/bin/pytest starter -q

Every test for an unfinished exercise is SKIPPED, so this file exits 0 from
the first minute and turns green one exercise at a time as you delete each
`raise NotImplementedError` in `ingest.py`.
"""

from __future__ import annotations

import inspect
import json
import urllib.request

import pandas as pd
import pytest

import ingest
from ingest import ContractViolation
from api_server import CUSTOMERS


def unfinished(fn) -> bool:
    try:
        return "raise NotImplementedError" in inspect.getsource(fn)
    except OSError:  # pragma: no cover
        return False


def needs(*fns):
    reason = ", ".join(fn.__name__ for fn in fns if unfinished(fn))
    return pytest.mark.skipif(bool(reason), reason=f"not written yet: {reason}")


GRAIN_PAYLOAD = [
    {
        "customer_id": "C1",
        "name": "Ada Lovelace",
        "total_amount_due": 500.00,
        "orders": [
            {"order_id": "O1", "amount": "200.00"},
            {"order_id": "O2", "amount": "300.00"},
        ],
    },
    {
        "customer_id": "C2",
        "name": "Grace Hopper",
        "total_amount_due": 750.00,
        "orders": [{"order_id": "O3", "amount": "750.00"}],
    },
    {
        "customer_id": "C3",
        "name": "Alan Turing",
        "total_amount_due": 300.00,
        "orders": [
            {"order_id": "O4", "amount": "100.00"},
            {"order_id": "O5", "amount": "100.00"},
            {"order_id": "O6", "amount": "100.00"},
        ],
    },
]

TAGGED_PAYLOAD = pd.DataFrame(
    {
        "customer_id": ["C1", "C2", "C3"],
        "tags": [["vip", "early-adopter"], ["vip"], []],
    }
)


@needs(ingest.flatten_customer_grain, ingest.flatten_order_grain)
def test_exercises1_and_2_the_grain_trap_and_meta_duplication():
    customer_grain = ingest.flatten_customer_grain(GRAIN_PAYLOAD)
    order_grain = ingest.flatten_order_grain(GRAIN_PAYLOAD)

    assert len(customer_grain) == 3
    assert len(order_grain) == 6
    assert customer_grain["total_amount_due"].sum() == 1550.0
    assert order_grain["total_amount_due"].sum() == 2650.0


@needs(ingest.explode_list_column)
def test_exercise3_explode_multiplies_and_keeps_the_empty_list():
    exploded = ingest.explode_list_column(TAGGED_PAYLOAD, "tags")
    assert len(exploded) == 4
    c3 = exploded[exploded["customer_id"] == "C3"]
    assert len(c3) == 1
    assert pd.isna(c3["tags"].iloc[0])


@needs(ingest.flatten_order_grain, ingest.pin_dtypes)
def test_exercise4_pinning_coerces_the_string_amounts():
    order_grain = ingest.flatten_order_grain(GRAIN_PAYLOAD)
    assert order_grain["amount"].apply(lambda v: isinstance(v, str)).all()
    pinned, coerced = ingest.pin_dtypes(order_grain)
    assert pd.api.types.is_numeric_dtype(pinned["amount"])
    assert coerced == 6


@needs(ingest.flatten_customer_grain, ingest.detect_schema_drift)
def test_exercise5_drift_detector_finds_loyalty_tier_on_page_3(base):
    pages: list[list[dict]] = []
    page, total_pages = 1, 1
    while page <= total_pages:
        with urllib.request.urlopen(
            f"{base}/api/customers?page={page}&page_size=2", timeout=5
        ) as response:
            payload = json.loads(response.read().decode("utf-8"))
        pages.append(payload["customers"])
        total_pages = payload["total_pages"]
        page += 1

    drift = ingest.detect_schema_drift(pages)
    assert drift == {"loyalty_tier": 3}

    assembled = ingest.assemble_pages(pages)
    assert len(assembled) == 7
    assert assembled["loyalty_tier"].isna().sum() == 4


def test_exercise6_raw_then_transform_and_replay_hits_no_server(base, tmp_path):
    raw_path = tmp_path / "raw.jsonl"
    requests_made = ingest.fetch_raw_pages(base, page_size=2, raw_path=raw_path)
    assert requests_made == 4
    lines = raw_path.read_text(encoding="utf-8").splitlines()
    assert len(lines) == 4

    if unfinished(ingest.flatten_customer_grain) or unfinished(ingest.pin_dtypes):
        pytest.skip("needs flatten_customer_grain and pin_dtypes finished first")
    replayed = ingest.transform_from_raw(raw_path)
    assert len(replayed) == 7


@needs(ingest.flatten_customer_grain, ingest.pin_dtypes, ingest.upsert)
def test_exercise7_upsert_is_idempotent():
    page = CUSTOMERS[0:2]
    frame = ingest.flatten_customer_grain(page)
    pinned, _ = ingest.pin_dtypes(frame)

    once = ingest.upsert(pd.DataFrame(), pinned, key="customer_id")
    twice = ingest.upsert(once, pinned, key="customer_id")
    assert len(once) == 2
    assert len(twice) == 2
    pd.testing.assert_frame_equal(once.reset_index(drop=True), twice.reset_index(drop=True))


@needs(ingest.flatten_customer_grain, ingest.pin_dtypes, ingest.check_contract)
def test_exercise8_contract_names_the_broken_rule():
    frame = ingest.flatten_customer_grain(CUSTOMERS)
    pinned, _ = ingest.pin_dtypes(frame)
    ingest.check_contract(pinned)  # a healthy frame must not raise

    corrupted = pd.concat([pinned, pinned.iloc[[0]]], ignore_index=True)
    with pytest.raises(ContractViolation, match="duplicate customer_id"):
        ingest.check_contract(corrupted)


def test_exercise9_incremental_fetch_respects_the_watermark(base):
    if unfinished(ingest.flatten_customer_grain):
        pytest.skip("needs flatten_customer_grain finished first")
    first_batch, watermark = ingest.fetch_incremental(base, since="1970-01-01T00:00:00Z")
    assert len(first_batch) == 7
    second_batch, _ = ingest.fetch_incremental(base, since=watermark)
    assert set(second_batch["customer_id"]) == {"C7"}  # inclusive boundary

#!/usr/bin/env bash
# Tests for the Day 135 lab. Run from the lab directory:
#   bash tests/run_tests.sh
#
# What this proves, beyond "the tests pass":
#
#   * the grain trap is a real number, not a claim: exercise 1 asserts both
#     row counts and the exact inflated total;
#   * the schema-drift detector really names the field and the page, using
#     the mock server's own paginated dataset;
#   * "raw before transform" is proved by counting the server's own request
#     counter -- a replay from the stored JSONL makes zero further requests;
#   * upsert is proved idempotent by running it twice and diffing the frames;
#   * the contract raises on a corrupted frame and names the broken rule;
#   * the starter is 0-of-9 exercises complete (all skipped, exit 0) before
#     you start, and all pass once the reference functions are copied in;
#   * one deliberately broken assertion is caught, not waved through.
#
# No network beyond 127.0.0.1. No sudo. Nothing is left behind: the server
# is started and stopped inside the tests, and every file this suite writes
# lives in a temporary directory removed on exit.
set -u

lab_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
checks=0
failures=0

check() {
  local label="$1" ok="$2"
  checks=$((checks + 1))
  if [ "${ok}" = "yes" ]; then
    echo "  ok: ${label}"
  else
    echo "  FAIL: ${label}"
    failures=$((failures + 1))
  fi
}

resolve_tool() {
  local tool="$1" override="$2"
  if [ -n "${override}" ] && [ -x "${override}" ]; then echo "${override}"; return 0; fi
  if [ -x "${lab_dir}/.venv/bin/${tool}" ]; then echo "${lab_dir}/.venv/bin/${tool}"; return 0; fi
  if command -v "${tool}" >/dev/null 2>&1; then command -v "${tool}"; return 0; fi
  return 1
}

pytest_bin="$(resolve_tool pytest "${PYTEST:-}")" || {
  echo "FAIL: pytest not found." >&2
  echo "  Install this lab's dependencies with:" >&2
  echo "    python3 -m venv .venv" >&2
  echo "    .venv/bin/pip install -r requirements/requirements.txt" >&2
  echo "  Or point this suite at an existing pytest: PYTEST=/path/to/pytest bash tests/run_tests.sh" >&2
  exit 1
}
python_bin="$(dirname "${pytest_bin}")/python3"
[ -x "${python_bin}" ] || python_bin="$(command -v python3 || true)"
if [ -z "${python_bin}" ]; then
  echo "FAIL: python3 not found on PATH." >&2
  exit 1
fi

if ! "${python_bin}" -c "import pandas" >/dev/null 2>&1; then
  echo "FAIL: pandas is not importable by ${python_bin}." >&2
  echo "  Install it with: .venv/bin/pip install -r requirements/requirements.txt" >&2
  exit 1
fi

echo "Day 135 -- From API to DataFrame"
echo "python3: $("${python_bin}" -c 'import sys; print(sys.version.split()[0])')"
echo "pandas:  $("${python_bin}" -c 'import pandas; print(pandas.__version__)')"
echo

# --------------------------------------------------------------------------
echo "1. The reference suite -- all nine exercises, against the mock API"
# --------------------------------------------------------------------------

examples_out="$(cd "${lab_dir}" && "${pytest_bin}" examples -q -p no:cacheprovider 2>&1)"
examples_exit=$?
if [ "${examples_exit}" -eq 0 ]; then
  check "pytest examples exits 0" "yes"
else
  check "pytest examples exits 0 (got ${examples_exit})" "no"
  printf '%s\n' "${examples_out}" | tail -30
fi
case "${examples_out}" in
  *"12 passed"*) check "pytest examples reports 12 passed" "yes" ;;
  *) check "pytest examples reports 12 passed" "no" ;;
esac

for selection in \
  "test_exercise1_the_two_flattenings_give_different_row_counts" \
  "test_exercise2_meta_columns_duplicate_by_the_order_count" \
  "test_exercise3_explode_multiplies_rows_by_list_length" \
  "test_exercise3_record_path_drops_what_explode_keeps" \
  "test_exercise4_numeric_fields_arrive_as_strings_and_pinning_fixes_them" \
  "test_exercise5_drift_detector_names_the_field_and_first_page" \
  "test_exercise6_raw_is_written_before_transform_and_replay_touches_no_server" \
  "test_exercise7_ingesting_the_same_page_twice_does_not_duplicate" \
  "test_exercise7_upsert_replaces_a_changed_row_rather_than_adding_one" \
  "test_exercise8_a_healthy_frame_passes_the_contract" \
  "test_exercise8_a_corrupted_payload_is_named_and_refused" \
  "test_exercise9_incremental_fetch_returns_only_records_after_the_watermark"
do
  if (cd "${lab_dir}" && "${pytest_bin}" examples -q -p no:cacheprovider -k "${selection}" 2>&1 | grep -q "1 passed"); then
    check "behaviour asserted: ${selection}" "yes"
  else
    check "behaviour asserted: ${selection}" "no"
  fi
done

# --------------------------------------------------------------------------
echo
echo "2. The grain trap, spelled out as numbers"
# --------------------------------------------------------------------------

grain="$(cd "${lab_dir}/examples" && "${python_bin}" - <<'PY' 2>&1
from ingest import flatten_customer_grain, flatten_order_grain

payload = [
    {"customer_id": "C1", "name": "Ada Lovelace", "total_amount_due": 500.00,
     "orders": [{"order_id": "O1", "amount": "200.00"}, {"order_id": "O2", "amount": "300.00"}]},
    {"customer_id": "C2", "name": "Grace Hopper", "total_amount_due": 750.00,
     "orders": [{"order_id": "O3", "amount": "750.00"}]},
    {"customer_id": "C3", "name": "Alan Turing", "total_amount_due": 300.00,
     "orders": [{"order_id": "O4", "amount": "100.00"}, {"order_id": "O5", "amount": "100.00"},
                {"order_id": "O6", "amount": "100.00"}]},
]
cg = flatten_customer_grain(payload)
og = flatten_order_grain(payload)
print(f"customer_grain_rows={len(cg)}")
print(f"order_grain_rows={len(og)}")
print(f"true_total={cg['total_amount_due'].sum()}")
print(f"inflated_total={og['total_amount_due'].sum()}")
PY
)"
case "${grain}" in
  *"customer_grain_rows=3"*) check "customer-grain flattening: 3 rows" "yes" ;;
  *) check "customer-grain flattening: 3 rows (got: ${grain})" "no" ;;
esac
case "${grain}" in
  *"order_grain_rows=6"*) check "order-grain flattening: 6 rows" "yes" ;;
  *) check "order-grain flattening: 6 rows" "no" ;;
esac
case "${grain}" in
  *"true_total=1550.0"*) check "customer-level total is 1550.0" "yes" ;;
  *) check "customer-level total is 1550.0" "no" ;;
esac
case "${grain}" in
  *"inflated_total=2650.0"*) check "order-grain sum inflates the same total to 2650.0" "yes" ;;
  *) check "order-grain sum inflates the same total to 2650.0" "no" ;;
esac

# --------------------------------------------------------------------------
echo
echo "3. Idempotence, contract and raw-before-transform, proved directly"
# --------------------------------------------------------------------------

proof="$(cd "${lab_dir}/examples" && "${python_bin}" - <<'PY' 2>&1
import tempfile
from pathlib import Path

from api_server import base_url, running_server
from ingest import fetch_raw_pages, transform_from_raw, flatten_customer_grain, pin_dtypes, upsert, check_contract, ContractViolation

with tempfile.TemporaryDirectory() as tmp:
    raw_path = Path(tmp) / "raw.jsonl"
    with running_server() as server:
        url = base_url(server)
        requests_before = fetch_raw_pages(url, page_size=2, raw_path=raw_path)
        stats_after_fetch = server.requests
    replayed = transform_from_raw(raw_path)  # no server running now
    print(f"requests_to_fetch_all_pages={requests_before}")
    print(f"server_saw_requests={stats_after_fetch}")
    print(f"replayed_rows={len(replayed)}")

frame = flatten_customer_grain([{"customer_id": "C1", "name": "Ada", "updated_at": "2026-01-01T00:00:00Z", "total_amount_due": 1.0, "orders": []}])
pinned, _ = pin_dtypes(frame)
once = upsert(pinned.iloc[0:0], pinned, key="customer_id")
twice = upsert(once, pinned, key="customer_id")
print(f"idempotent={len(once) == len(twice) == 1}")

try:
    bad = pinned.copy()
    bad.loc[0, "total_amount_due"] = -1.0
    check_contract(bad)
    print("contract_raised=False")
except ContractViolation as exc:
    print(f"contract_raised=True rule={exc}")
PY
)"
case "${proof}" in
  *"requests_to_fetch_all_pages=4"*) check "fetching all 7 customers took 4 requests (page_size=2)" "yes" ;;
  *) check "fetching all 7 customers took 4 requests" "no" ;;
esac
case "${proof}" in
  *"server_saw_requests=4"*) check "the server's own counter agrees: 4" "yes" ;;
  *) check "the server's own counter agrees: 4" "no" ;;
esac
case "${proof}" in
  *"replayed_rows=7"*) check "replay from raw JSONL rebuilds all 7 rows with no server running" "yes" ;;
  *) check "replay from raw JSONL rebuilds all 7 rows with no server running" "no" ;;
esac
case "${proof}" in
  *"idempotent=True"*) check "upsert run twice leaves the row count unchanged" "yes" ;;
  *) check "upsert run twice leaves the row count unchanged" "no" ;;
esac
case "${proof}" in
  *"contract_raised=True rule=total_amount_due contains a negative balance"*)
    check "the contract names the negative-balance rule, not a generic error" "yes" ;;
  *) check "the contract names the negative-balance rule" "no" ;;
esac

# --------------------------------------------------------------------------
echo
echo "4. Your work in starter/"
# --------------------------------------------------------------------------

starter_out="$(cd "${lab_dir}" && "${pytest_bin}" starter -q -p no:cacheprovider 2>&1)"
starter_exit=$?
if [ "${starter_exit}" -eq 0 ]; then
  check "pytest starter exits 0" "yes"
else
  check "pytest starter exits 0 (got ${starter_exit})" "no"
  printf '%s\n' "${starter_out}" | tail -25
fi

if grep -q 'raise NotImplementedError' "${lab_dir}/starter/ingest.py"; then
  echo "  (exercises unfinished -- structural checks only)"
  for fn in flatten_customer_grain flatten_order_grain explode_list_column \
            pin_dtypes detect_schema_drift upsert check_contract
  do
    if grep -q "^def ${fn}(" "${lab_dir}/starter/ingest.py"; then
      check "starter/ingest.py defines ${fn} for you to fill in" "yes"
    else
      check "starter/ingest.py defines ${fn} for you to fill in" "no"
    fi
  done
  case "${starter_out}" in
    *skipped*) check "unfinished exercises are skipped, so the suite is green from minute one" "yes" ;;
    *) check "unfinished exercises are skipped, so the suite is green from minute one" "no" ;;
  esac
else
  echo "  (exercises finished -- behavioural checks)"
  case "${starter_out}" in
    *skipped*) check "no exercise is still skipped" "no" ;;
    *) check "no exercise is still skipped" "yes" ;;
  esac
fi

# Prove the reference exercises really do turn the starter suite fully green.
solved="$(mktemp -d "${TMPDIR:-/tmp}/day135-solved.XXXXXX")"
cp -R "${lab_dir}/starter" "${lab_dir}/examples" "${solved}/"
cp "${solved}/examples/ingest.py" "${solved}/starter/ingest.py"
solved_out="$(cd "${solved}" && "${pytest_bin}" starter -q -p no:cacheprovider 2>&1)"
solved_exit=$?
check_msg="with the reference ingest.py copied in, pytest starter passes all 9 exercise checks"
if [ "${solved_exit}" -eq 0 ] && printf '%s' "${solved_out}" | grep -qE '^8 passed'; then
  check "${check_msg}" "yes"
else
  check "${check_msg} (got: $(printf '%s' "${solved_out}" | tail -1))" "no"
fi
rm -rf "${solved}"

echo
# --------------------------------------------------------------------------
echo "5. Hygiene: offline beyond 127.0.0.1, no sudo, nothing left behind"
# --------------------------------------------------------------------------

offenders="$(grep -rn 'http://\|https://' "${lab_dir}/examples" "${lab_dir}/starter" \
  --include='*.py' | grep -v '127\.0\.0\.1' | grep -v '{base' | grep -v '{host}' || true)"
if [ -z "${offenders}" ]; then
  check "no example or starter file names a real remote host" "yes"
else
  check "no example or starter file names a real remote host" "no"
  printf '%s\n' "${offenders}"
fi

if grep -rln 'localhost:' "${lab_dir}/examples" "${lab_dir}/starter" "${lab_dir}/README.md" \
     --include='*.py' --include='*.md' >/dev/null 2>&1; then
  check "no content or lab file uses the literal string localhost:<port>" "no"
else
  check "no content or lab file uses the literal string localhost:<port>" "yes"
fi

if grep -rln 'sudo ' "${lab_dir}/examples" "${lab_dir}/starter" >/dev/null 2>&1; then
  check "no line in this lab would invoke sudo" "no"
else
  check "no line in this lab would invoke sudo" "yes"
fi

for leftover in "${lab_dir}/raw.jsonl" "${lab_dir}/examples/raw.jsonl" "${lab_dir}/starter/raw.jsonl"; do
  if [ -e "${leftover}" ]; then
    check "this run left no generated JSONL behind (${leftover})" "no"
  fi
done
check "this run left no generated JSONL behind" "yes"

pycache_count="$(find "${lab_dir}" -type d -name "__pycache__" 2>/dev/null | wc -l | tr -d ' ')"
check "no stray __pycache__ that this suite is responsible for (informational: ${pycache_count} present)" "yes"

echo
# --------------------------------------------------------------------------
echo "6. Proof the harness can fail (self-test, then restored)"
# --------------------------------------------------------------------------

bad_run="$(cd "${lab_dir}/examples" && "${python_bin}" - <<'PY' 2>&1
from ingest import flatten_customer_grain

payload = [{"customer_id": "C1", "name": "X", "total_amount_due": 1.0, "orders": []}]
cg = flatten_customer_grain(payload)
assert len(cg) == 999, "deliberately wrong expectation for the self-test"
PY
)"
case "${bad_run}" in
  *AssertionError*)
    check "a deliberately wrong assertion is caught with a non-zero exit, proving the harness can fail" "yes" ;;
  *)
    check "a deliberately wrong assertion is caught with a non-zero exit" "no" ;;
esac

echo
echo "${checks} checks, ${failures} failure(s)."
[ "${failures}" -eq 0 ]