Hands-on lab — Day 94: Data Validation with pydantic

[
  {
    "reading_id": "RD-0001",
    "station": { "code": "ST-KLM", "name": "Kalmar Ridge", "elevation_m": 340 },
    "recorded_at": "2026-08-15T06:00:00Z",
    "pm2_5": 12.4,
    "temperature_c": 18.2,
    "humidity_pct": 61,
    "operator": "R. Nayar",
    "notes": null
  },
  {
    "reading_id": "RD-0002",
    "station": { "code": "ST-KLM", "name": "  Kalmar Ridge  ", "elevation_m": "340" },
    "recorded_at": "2026-08-15T07:00:00Z",
    "pm2_5": "14.8",
    "temperature_c": "19",
    "humidity_pct": "58",
    "operator": "R. Nayar",
    "notes": "  routine sweep  "
  },
  {
    "reading_id": "RD-0003",
    "station": { "code": "ST-BRW", "name": "Brantwood Flats", "elevation_m": 88 },
    "recorded_at": "2026-08-15T07:00:00Z",
    "pm2_5": 9.1,
    "temperature_c": 21.0,
    "humidity_pct": 54,
    "notes": "operator field never written by the old exporter"
  },
  {
    "reading_id": "RD-0004",
    "station": { "code": "ST-BRW", "name": "Brantwood Flats", "elevation_m": 88 },
    "recorded_at": "2026-08-15T08:00:00Z",
    "pm2_5": "not-measured",
    "temperature_c": 22.3,
    "humidity_pct": 52,
    "operator": "T. Oyelaran",
    "notes": null
  },
  {
    "reading_id": "RD-0005",
    "station": { "code": "ST-KLM", "name": "Kalmar Ridge", "elevation_m": 340 },
    "recorded_at": "2026-08-15T08:00:00Z",
    "pm2_5": 11.7,
    "temperature_c": 19.4,
    "humidity_pct": 118,
    "operator": "R. Nayar",
    "notes": null
  },
  {
    "reading_id": "RD-0006",
    "station": { "code": "ST-HVN", "name": "Havenmoor Pier", "elevation_m": 4 },
    "recorded_at": "2026-08-15T09:00:00Z",
    "pm2_5": 33.6,
    "temperature_c": 17.8,
    "humidty_pct": 71,
    "operator": "M. Ferreira",
    "notes": null
  },
  {
    "reading_id": "RD-0007",
    "station": { "code": "ST-north", "name": "Northgate Yard", "elevation_m": 210 },
    "recorded_at": "2026-08-15T09:00:00Z",
    "pm2_5": 15.2,
    "temperature_c": 18.9,
    "humidity_pct": 63,
    "operator": "M. Ferreira",
    "notes": null
  },
  {
    "reading_id": "RD-0001",
    "station": { "code": "ST-HVN", "name": "Havenmoor Pier", "elevation_m": 4 },
    "recorded_at": "2026-08-15T10:00:00Z",
    "pm2_5": 16.0,
    "temperature_c": 18.5,
    "humidity_pct": 66,
    "operator": "M. Ferreira",
    "notes": "re-used id from a hand-edited export"
  },
  {
    "reading_id": "RD-0009",
    "station": { "code": "ST-BRW", "name": "Brantwood Flats", "elevation_m": 88 },
    "recorded_at": "15/08/2026 10:00",
    "pm2_5": 10.3,
    "temperature_c": 23.1,
    "humidity_pct": 49,
    "operator": "T. Oyelaran",
    "notes": null
  },
  {
    "reading_id": "RD-0010",
    "station": { "code": "ST-HVN", "name": "Havenmoor Pier", "elevation_m": 4 },
    "recorded_at": "2026-08-15T11:00:00Z",
    "pm2_5": 612.5,
    "temperature_c": 31.7,
    "humidity_pct": 44,
    "operator": "M. Ferreira",
    "notes": null
  },
  {
    "reading_id": "RD-0011",
    "station": { "code": "ST-HVN", "name": "Havenmoor Pier", "elevation_m": 4 },
    "recorded_at": "2026-08-15T11:00:00Z",
    "pm2_5": 548.9,
    "temperature_c": 30.4,
    "humidity_pct": 46,
    "operator": "M. Ferreira",
    "notes": "smoke plume from the north, confirmed by the duty log"
  },
  {
    "reading_id": "RD-0012",
    "station": { "code": "ST-KLM", "name": "Kalmar Ridge", "elevation_m": 340 },
    "recorded_at": "2026-08-15T12:00:00+02:00",
    "pm2_5": 8.05,
    "temperature_c": -3.5,
    "humidity_pct": 0,
    "operator": null,
    "notes": null
  }
]

"""Which conversions pydantic performs by default, and which it refuses.

This is the part of pydantic that surprises people, so rather than describe it,
this module asks. Every row of the table it prints is the result of an actual
call to ``TypeAdapter(...).validate_python(...)`` in this interpreter, once in
the default (lax) mode and once with ``strict=True``.

Run it directly:

    python3 examples/coercion.py
"""

from __future__ import annotations

from datetime import date, datetime
from typing import Any, NamedTuple, get_origin

from pydantic import TypeAdapter, ValidationError

CASES: list[tuple[Any, Any]] = [
    ("42", int),
    ("42.0", int),
    ("  42  ", int),
    ("forty-two", int),
    (42.0, int),
    (42.7, int),
    (True, int),
    ("3.14", float),
    (3, float),
    (42, str),
    (None, str),
    ("yes", bool),
    ("true", bool),
    (1, bool),
    ("2026-08-15T06:00:00Z", datetime),
    ("15/08/2026", date),
    (1786773600, datetime),
    ("[1, 2]", list[int]),
    ((1, 2), list[int]),
    # A set of small ints, deliberately: CPython hash-randomises strings, so a
    # set of strings would print in a different order on a different run and
    # this table would stop being reproducible.
    ({1, 2}, list[int]),
]


class Row(NamedTuple):
    """One question asked twice."""

    value: Any
    target: Any
    lax: str
    strict: str


def _target_name(target: Any) -> str:
    if get_origin(target) is not None:
        return str(target).replace("typing.", "")
    return getattr(target, "__name__", None) or str(target)


def _ask(value: Any, target: Any, *, strict: bool) -> str:
    """Return either the validated value's repr or the error ``type``."""
    adapter = TypeAdapter(target)
    try:
        return repr(adapter.validate_python(value, strict=strict))
    except ValidationError as exc:
        return f"refused: {exc.errors()[0]['type']}"


def coercion_table() -> list[Row]:
    rows = []
    for value, target in CASES:
        rows.append(
            Row(
                value=value,
                target=target,
                lax=_ask(value, target, strict=False),
                strict=_ask(value, target, strict=True),
            )
        )
    return rows


def render(rows: list[Row]) -> str:
    header = f"{'input':<24} {'declared as':<12} {'lax (default)':<34} strict=True"
    lines = [header, "-" * len(header) + "----"]
    for row in rows:
        # A set has no defined order, so print it in a stable form.
        shown = sorted(row.value, key=str) if isinstance(row.value, set) else row.value
        prefix = "set" if isinstance(row.value, set) else ""
        lines.append(
            f"{prefix + repr(shown):<24} {_target_name(row.target):<12} {row.lax:<34} {row.strict}"
        )
    return "\n".join(lines)


def main() -> int:
    print(render(coercion_table()))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""The data-quality gate: validate a whole batch, keep the good, report the bad.

The rule this module exists to enforce is simple to state and easy to get
wrong: **one bad record must not end the run.** A pipeline that raises on the
first malformed row processes nothing and tells you about one problem. A
pipeline with a gate processes everything it can, and hands back a report
naming every record it refused and exactly why.

Run it directly:

    python3 examples/gate.py
    python3 examples/gate.py --input data/raw-readings.json --out-dir out
"""

from __future__ import annotations

import argparse
import json
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from pydantic import ValidationError

from models import Reading

LAB_DIR = Path(__file__).resolve().parent.parent
DEFAULT_INPUT = LAB_DIR / "data" / "raw-readings.json"
DEFAULT_OUT_DIR = LAB_DIR / "out"


@dataclass(frozen=True)
class Rejection:
    """One record the gate refused, and every reason it refused it."""

    index: int
    reading_id: str | None
    errors: list[dict[str, Any]]

    def summary(self) -> str:
        parts = []
        for error in self.errors:
            where = ".".join(str(part) for part in error["loc"]) or "<record>"
            parts.append(f"{where} [{error['type']}]")
        label = self.reading_id or "<no id>"
        return f"record {self.index} ({label}): " + "; ".join(parts)


@dataclass
class GateResult:
    """What came out of the gate."""

    accepted: list[Reading] = field(default_factory=list)
    rejected: list[Rejection] = field(default_factory=list)

    @property
    def seen(self) -> int:
        return len(self.accepted) + len(self.rejected)

    @property
    def accepted_count(self) -> int:
        return len(self.accepted)

    @property
    def rejected_count(self) -> int:
        return len(self.rejected)

    def error_types(self) -> list[str]:
        return [error["type"] for rejection in self.rejected for error in rejection.errors]

    def as_report(self) -> dict[str, Any]:
        """A machine-readable report. Whoever owns the source data reads this."""
        return {
            "records_seen": self.seen,
            "records_accepted": self.accepted_count,
            "records_rejected": self.rejected_count,
            "rejections": [
                {
                    "index": rejection.index,
                    "reading_id": rejection.reading_id,
                    "errors": rejection.errors,
                }
                for rejection in self.rejected
            ],
        }


def _tidy(errors: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Keep the four fields that matter and make ``loc`` JSON-friendly.

    ``msg`` is kept because a human has to read the report. It is deliberately
    the only one of the four that no test in this lab asserts on.
    """
    tidy = []
    for error in errors:
        tidy.append(
            {
                "loc": list(error["loc"]),
                "type": error["type"],
                "msg": error["msg"],
                "input": _jsonable(error.get("input")),
            }
        )
    return tidy


def _jsonable(value: Any) -> Any:
    try:
        json.dumps(value)
    except (TypeError, ValueError):
        return repr(value)
    return value


def load_records(path: Path = DEFAULT_INPUT) -> list[Any]:
    """Read the raw batch. Note what this does *not* do: it does not validate."""
    with path.open(encoding="utf-8") as handle:
        data = json.load(handle)
    if not isinstance(data, list):
        raise ValueError(f"{path} must contain a JSON array of records")
    return data


def run_gate(records: list[Any]) -> GateResult:
    """Validate every record. Never raises for bad data — that is the whole point."""
    result = GateResult()
    seen_ids: dict[str, int] = {}

    for index, raw in enumerate(records):
        raw_id = raw.get("reading_id") if isinstance(raw, dict) else None
        label = raw_id if isinstance(raw_id, str) else None

        try:
            reading = Reading.model_validate(raw)
        except ValidationError as exc:
            result.rejected.append(
                Rejection(index=index, reading_id=label, errors=_tidy(exc.errors()))
            )
            continue

        # A batch-level rule. No per-record schema can express it, because
        # uniqueness is a property of the batch, not of the record.
        first_seen = seen_ids.get(reading.reading_id)
        if first_seen is not None:
            result.rejected.append(
                Rejection(
                    index=index,
                    reading_id=reading.reading_id,
                    errors=[
                        {
                            "loc": ["reading_id"],
                            "type": "duplicate_id",
                            "msg": f"reading_id already used by record {first_seen}",
                            "input": reading.reading_id,
                        }
                    ],
                )
            )
            continue

        seen_ids[reading.reading_id] = index
        result.accepted.append(reading)

    return result


def write_outputs(result: GateResult, out_dir: Path = DEFAULT_OUT_DIR) -> tuple[Path, Path]:
    """Emit the accepted records and the rejection report as two files."""
    out_dir.mkdir(parents=True, exist_ok=True)
    accepted_path = out_dir / "accepted.jsonl"
    report_path = out_dir / "rejects.json"

    with accepted_path.open("w", encoding="utf-8") as handle:
        for reading in result.accepted:
            handle.write(reading.model_dump_json() + "\n")

    report_path.write_text(
        json.dumps(result.as_report(), indent=2, sort_keys=False) + "\n",
        encoding="utf-8",
    )
    return accepted_path, report_path


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Validate a batch of sensor readings.")
    parser.add_argument("--input", type=Path, default=DEFAULT_INPUT)
    parser.add_argument("--out-dir", type=Path, default=DEFAULT_OUT_DIR)
    parser.add_argument(
        "--fail-over",
        type=float,
        default=None,
        metavar="FRACTION",
        help="exit non-zero if more than this fraction of records were rejected",
    )
    args = parser.parse_args(argv)

    records = load_records(args.input)
    result = run_gate(records)
    accepted_path, report_path = write_outputs(result, args.out_dir)

    print(f"read      {result.seen} records from {args.input.name}")
    print(f"accepted  {result.accepted_count}")
    print(f"rejected  {result.rejected_count}")
    print()
    for rejection in result.rejected:
        print("  " + rejection.summary())
    print()
    try:
        shown = args.out_dir.resolve().relative_to(LAB_DIR)
    except ValueError:
        shown = args.out_dir
    print(f"wrote {accepted_path.name} and {report_path.name} to {shown}/")

    if args.fail_over is not None and result.seen:
        share = result.rejected_count / result.seen
        if share > args.fail_over:
            print(
                f"\nreject rate {share:.1%} exceeds the {args.fail_over:.0%} threshold",
                file=sys.stderr,
            )
            return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""The boundary schema for the air-quality feed, written with pydantic v2.

Every record that enters the pipeline is checked against ``Reading`` before any
other code is allowed to see it. If a record does not satisfy this file, it does
not exist as far as the rest of the program is concerned.

All names, station codes and operator initials in the sample data are invented
for this lab. No real monitoring network, person or measurement is involved.
"""

from __future__ import annotations

from datetime import datetime
from typing import Annotated

from pydantic import (
    BaseModel,
    ConfigDict,
    Field,
    StringConstraints,
    computed_field,
    field_validator,
    model_validator,
)

__all__ = ["Reading", "Station", "StationCode", "Percent", "Micrograms"]


# --------------------------------------------------------------------------
# Annotated types: say what a constrained value *is*, once, then reuse it.
# --------------------------------------------------------------------------

StationCode = Annotated[str, StringConstraints(pattern=r"^ST-[A-Z]{3}$")]
"""A station code: the letters ``ST-`` and exactly three capitals, e.g. ST-KLM."""

Percent = Annotated[int, Field(ge=0, le=100)]
"""A whole percentage. 0 and 100 are both legal; 101 is not."""

Micrograms = Annotated[float, Field(ge=0.0, le=1000.0)]
"""Micrograms per cubic metre. Negative is impossible; 1000 is the sensor ceiling."""

ReadingId = Annotated[str, StringConstraints(pattern=r"^RD-\d{4}$")]
"""A reading id: ``RD-`` and exactly four digits."""


# --------------------------------------------------------------------------
# The models
# --------------------------------------------------------------------------


class Station(BaseModel):
    """Where a reading was taken. A nested model, validated in its own right."""

    model_config = ConfigDict(
        extra="forbid",
        str_strip_whitespace=True,
        frozen=True,
    )

    code: StationCode
    name: str = Field(min_length=1, max_length=60, description="Human-readable site name")
    elevation_m: int = Field(ge=-500, le=9000, description="Metres above sea level")


class Reading(BaseModel):
    """One measurement from one station at one moment."""

    model_config = ConfigDict(
        extra="forbid",
        str_strip_whitespace=True,
        validate_assignment=True,
        populate_by_name=True,
    )

    reading_id: ReadingId
    station: Station
    recorded_at: datetime = Field(description="ISO 8601 timestamp, timezone required")

    # The vendor's export writes ``pm2_5``; the rest of this codebase says
    # ``pm25``. The alias is where that difference is absorbed, once.
    pm25: Micrograms = Field(alias="pm2_5", description="PM2.5 concentration")

    temperature_c: float = Field(ge=-90.0, le=60.0)
    humidity_pct: Percent

    # Required, and may be null. Three different things live in these two lines:
    #   * ``operator`` is REQUIRED — the key must be present — and NULLABLE.
    #   * ``notes`` is OPTIONAL — it has a default — and also nullable.
    # A field can be optional and not nullable, or nullable and not optional.
    operator: str | None
    notes: str | None = None

    @field_validator("notes", mode="after")
    @classmethod
    def blank_note_is_no_note(cls, value: str | None) -> str | None:
        """An empty string is not a note. Normalise it away before it spreads."""
        if value is not None and value.strip() == "":
            return None
        return value

    @field_validator("recorded_at", mode="after")
    @classmethod
    def timestamp_must_carry_a_timezone(cls, value: datetime) -> datetime:
        """A naive timestamp is an ambiguity, not a time."""
        if value.tzinfo is None:
            raise ValueError("recorded_at must include a timezone offset")
        return value

    @model_validator(mode="after")
    def a_high_reading_must_be_explained(self) -> Reading:
        """A cross-field rule: no single-field constraint can express this.

        Above 500 micrograms the instrument is either witnessing something
        serious or misbehaving, and the difference is not in the number. The
        schema therefore demands that a human wrote down which it was.
        """
        if self.pm25 > 500.0 and not (self.notes and self.notes.strip()):
            raise ValueError("a pm25 reading above 500 requires a note explaining it")
        return self

    @computed_field  # type: ignore[prop-decorator]
    @property
    def band(self) -> str:
        """A derived label. Stored nowhere, serialised everywhere."""
        if self.pm25 <= 12.0:
            return "good"
        if self.pm25 <= 35.4:
            return "moderate"
        if self.pm25 <= 55.4:
            return "unhealthy-for-sensitive-groups"
        if self.pm25 <= 150.4:
            return "unhealthy"
        return "hazardous"

"""Run the miniature validator over the real batch, then compare it to pydantic.

The from-scratch validator in ``scratch_validator.py`` is about two hundred
lines and does five things: it finds the fields, it decides what "present"
means, it applies a small coercion policy, it recurses into nested models, and
it collects **every** error instead of raising on the first. That last one is
the part that looks easy and is not.

What this script exists to show is the gap. The toy and ``models.py`` describe
the same twelve records. The toy waves through nine of them; the pydantic
schema keeps five. (The gate in ``gate.py`` then drops one more, because a
duplicate id is a property of the batch and no per-record schema can see it.)
The difference is not that pydantic is stricter by temperament — it is that
ranges, patterns, timestamps, aliases and cross-field rules are things the toy
has no vocabulary for, and each one would have to be hand-written per field.
That is what "hand-written validation rots" means in practice.

Run it directly:

    python3 examples/scratch_demo.py
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from models import Reading
from pydantic import ValidationError
from scratch_models import ScratchReading
from scratch_validator import ValidationReport, format_report, validate

LAB_DIR = Path(__file__).resolve().parent.parent
BATCH = LAB_DIR / "data" / "raw-readings.json"


def load() -> list[Any]:
    with BATCH.open(encoding="utf-8") as handle:
        return json.load(handle)


def scratch_pass(records: list[Any]) -> tuple[int, list[tuple[int, list[dict[str, Any]]]]]:
    """Validate the batch with the hand-rolled validator. Returns (accepted, rejected)."""
    accepted = 0
    rejected: list[tuple[int, list[dict[str, Any]]]] = []
    for index, raw in enumerate(records):
        value, errors = validate(ScratchReading, raw)
        if errors:
            rejected.append((index, errors))
        else:
            accepted += 1
    return accepted, rejected


def pydantic_pass(records: list[Any]) -> tuple[int, list[tuple[int, list[dict[str, Any]]]]]:
    """The same batch through ``Reading``. Note it never raises out of this loop."""
    accepted = 0
    rejected: list[tuple[int, list[dict[str, Any]]]] = []
    for index, raw in enumerate(records):
        try:
            Reading.model_validate(raw)
        except ValidationError as exc:
            rejected.append((index, exc.errors()))
        else:
            accepted += 1
    return accepted, rejected


def _locs(errors: list[dict[str, Any]]) -> str:
    return ", ".join(
        f"{'.'.join(str(p) for p in error['loc']) or '<record>'} [{error['type']}]"
        for error in errors
    )


def main() -> int:
    records = load()

    print("=" * 74)
    print("1. One record, many problems at once")
    print("=" * 74)
    print()
    print("A validator that raises on the first bad field makes you fix a file one")
    print("round trip at a time. Both of these report everything they found.")
    print()

    # Deliberately broken in four separate ways, so the collecting behaviour is
    # visible rather than asserted. Every value here is invented for this lab.
    broken = {
        "reading_id": "RD-0099",
        "station": {"code": "ST-QQQ", "name": "Quarry Head", "elevation_m": "high"},
        "recorded_at": "2026-08-15T13:00:00Z",
        "pm2_5": "unreadable",
        "temperature_c": 20.0,
        "humidity_pct": None,
        "operator": "A. Invented",
    }

    _, scratch_errors = validate(ScratchReading, broken)
    print("from scratch:")
    print(format_report(scratch_errors))
    print()

    try:
        Reading.model_validate(broken)
    except ValidationError as exc:
        pydantic_errors = exc.errors()
    else:  # pragma: no cover - the record above cannot validate
        pydantic_errors = []
    print(f"pydantic: {len(pydantic_errors)} validation error(s)")
    for error in pydantic_errors:
        where = ".".join(str(part) for part in error["loc"]) or "<record>"
        print(f"  {where}")
        print(f"    type={error['type']} input={error['input']!r}")
    print()

    report = ValidationReport(scratch_errors)
    print(f"the toy's report answers questions: {len(report)} errors, types {report.types()}")
    print()

    print("=" * 74)
    print("2. The same twelve records through both")
    print("=" * 74)
    print()

    scratch_ok, scratch_bad = scratch_pass(records)
    pyd_ok, pyd_bad = pydantic_pass(records)

    print(f"from scratch : accepted {scratch_ok}, rejected {len(scratch_bad)}")
    for index, errors in scratch_bad:
        print(f"    record {index}: {_locs(errors)}")
    print()
    print(f"pydantic     : accepted {pyd_ok}, rejected {len(pyd_bad)}")
    for index, errors in pyd_bad:
        print(f"    record {index}: {_locs(errors)}")
    print()

    print("=" * 74)
    print("3. What the toy let through, and why")
    print("=" * 74)
    print()
    scratch_rejected = {index for index, _ in scratch_bad}
    for index, errors in pyd_bad:
        if index in scratch_rejected:
            continue
        reasons = ", ".join(sorted({error["type"] for error in errors}))
        print(f"  record {index}: pydantic says {reasons}; the toy has no rule for it")
    print()
    print("None of those are exotic. They are a range, a pattern, a date format and")
    print("a rule that spans two fields — and each one is a function the toy would")
    print("need hand-written, per field, and kept correct forever.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""The same two models expressed for the from-scratch validator.

Put this file beside ``models.py`` and read them together. The shapes are
identical; what differs is how much of "valid" each one can actually say.

The miniature validator understands types, presence, nullability, nesting and
unexpected keys. It has no vocabulary at all for ranges, patterns, lengths,
dates, aliases or cross-field rules — every one of those would have to be
hand-written per field, which is precisely how hand-written validation rots.
"""

from __future__ import annotations

from scratch_validator import MiniModel


class ScratchStation(MiniModel):
    code: str
    name: str
    elevation_m: int


class ScratchReading(MiniModel):
    reading_id: str
    station: ScratchStation
    recorded_at: str
    pm2_5: float
    temperature_c: float
    humidity_pct: int
    operator: str | None
    notes: str | None = None

"""A miniature validator built from first principles — no third-party code.

The point of this module is not to compete with pydantic. It is to make every
decision pydantic makes for you visible, by making you make it yourself:

  * where does the list of fields come from?          -> ``__annotations__``
  * what counts as "present"?                         -> a sentinel, not ``None``
  * which conversions are allowed?                    -> the COERCIONS table below
  * what happens on the first failure?                -> nothing; we keep going
  * how does a caller know *where* it went wrong?     -> a ``loc`` tuple

Everything here is the standard library. A model is an ordinary class whose
annotations describe its fields and whose class attributes supply defaults.

    class Station(MiniModel):
        code: str
        name: str
        elevation_m: int

    class Reading(MiniModel):
        reading_id: str
        station: Station
        pm25: float
        notes: str | None = None

    value, errors = validate(Reading, raw_dict)

``errors`` is a list of dictionaries, one per problem, each carrying ``loc``,
``type``, ``msg`` and ``input`` — the same four keys pydantic uses, chosen here
so that the shape of the report is familiar when you meet the real thing.
"""

from __future__ import annotations

import types
import typing
from typing import Any

__all__ = [
    "MISSING",
    "MiniModel",
    "ValidationReport",
    "coerce",
    "format_report",
    "validate",
]


class _Missing:
    """A sentinel meaning "the key was absent", which ``None`` cannot mean.

    This is the first decision a validator forces you to make. If absence were
    represented by ``None``, then a field explicitly set to ``null`` in the
    input would be indistinguishable from a field nobody wrote — and those are
    two different facts about the world.
    """

    __slots__ = ()

    def __repr__(self) -> str:  # pragma: no cover - debugging aid only
        return "MISSING"

    def __bool__(self) -> bool:
        return False


MISSING = _Missing()


class MiniModel:
    """Base class for a model. It carries no behaviour beyond field storage."""

    def __init__(self, **values: Any) -> None:
        for name, value in values.items():
            setattr(self, name, value)

    def __repr__(self) -> str:
        hints = declared_fields(type(self))
        pairs = ", ".join(f"{name}={getattr(self, name, MISSING)!r}" for name in hints)
        return f"{type(self).__name__}({pairs})"

    def __eq__(self, other: object) -> bool:
        if type(other) is not type(self):
            return NotImplemented
        return all(
            getattr(self, name, MISSING) == getattr(other, name, MISSING)
            for name in declared_fields(type(self))
        )

    def as_dict(self) -> dict[str, Any]:
        """The nearest thing this toy has to ``model_dump``."""
        out: dict[str, Any] = {}
        for name in declared_fields(type(self)):
            value = getattr(self, name, MISSING)
            out[name] = value.as_dict() if isinstance(value, MiniModel) else value
        return out


def declared_fields(model: type[MiniModel]) -> dict[str, Any]:
    """Resolve the annotations of ``model`` and its bases into {name: type}.

    ``typing.get_type_hints`` is used rather than reading ``__annotations__``
    directly, because under ``from __future__ import annotations`` every
    annotation is a *string* until something resolves it. That single detail is
    why hand-rolled validators so often work in one module and not in another.
    """
    hints = typing.get_type_hints(model)
    return {name: hint for name, hint in hints.items() if not name.startswith("_")}


def default_for(model: type[MiniModel], name: str) -> Any:
    """A class attribute of the same name is the field's default."""
    return getattr(model, name, MISSING)


# --------------------------------------------------------------------------
# The coercion policy — every entry here is a decision, not a law of nature.
# --------------------------------------------------------------------------


def _str_to_int(value: str) -> int:
    return int(value.strip())


def _str_to_float(value: str) -> float:
    return float(value.strip())


def _int_to_float(value: int) -> float:
    return float(value)


COERCIONS: dict[tuple[type, type], Any] = {
    (str, int): _str_to_int,
    (str, float): _str_to_float,
    (int, float): _int_to_float,
}
"""Which conversions this validator is willing to perform.

Deliberately absent, and each absence is a judgement:

``(float, int)``
    Refused because it loses information silently. ``3.7`` is not ``3``.
``(bool, int)``
    Refused because ``True`` really is an ``int`` in Python — ``isinstance(True,
    int)`` is ``True`` — and letting a checkbox arrive where a count was wanted
    is a bug that survives to production.
``(str, bool)``
    Refused because there is no single right answer. Is ``"0"`` false? Is
    ``"no"``? Every codebase picks differently, so this one picks nothing.
"""

TYPE_NAMES = {int: "int", float: "float", str: "string", bool: "bool"}


def coerce(value: Any, target: type) -> tuple[Any, str | None]:
    """Return ``(converted_value, error_type)``; ``error_type`` is ``None`` on success."""
    name = TYPE_NAMES.get(target, target.__name__)

    # ``bool`` is a subclass of ``int``. Check it before the isinstance below,
    # or every True in the input becomes a perfectly acceptable 1.
    if isinstance(value, bool) and target is not bool:
        return value, f"{name}_type"

    if type(value) is target:
        return value, None
    if isinstance(value, target) and not isinstance(value, bool):
        return value, None

    converter = COERCIONS.get((type(value), target))
    if converter is None:
        return value, f"{name}_type"
    try:
        return converter(value), None
    except (TypeError, ValueError):
        return value, f"{name}_parsing"


# --------------------------------------------------------------------------
# The validator proper
# --------------------------------------------------------------------------


def _optional_inner(hint: Any) -> tuple[Any, bool]:
    """Split ``T | None`` into ``(T, True)``; anything else into ``(hint, False)``."""
    origin = typing.get_origin(hint)
    if origin is types.UnionType or origin is typing.Union:
        args = [arg for arg in typing.get_args(hint) if arg is not type(None)]
        if len(args) == 1 and len(typing.get_args(hint)) == 2:
            return args[0], True
    return hint, False


def validate(
    model: type[MiniModel],
    raw: Any,
    *,
    loc: tuple[Any, ...] = (),
    allow_extra: bool = False,
) -> tuple[MiniModel | None, list[dict[str, Any]]]:
    """Validate ``raw`` against ``model``, collecting **every** problem.

    Returns ``(instance, [])`` on success and ``(None, errors)`` on failure.
    Never raises for bad data; a raise would end the run at the first problem,
    which is exactly the behaviour this module exists to avoid.
    """
    errors: list[dict[str, Any]] = []

    if not isinstance(raw, dict):
        return None, [
            {
                "loc": loc,
                "type": "model_type",
                "msg": f"Input should be an object for {model.__name__}",
                "input": raw,
            }
        ]

    fields = declared_fields(model)
    values: dict[str, Any] = {}

    for name, hint in fields.items():
        inner, nullable = _optional_inner(hint)
        field_loc = (*loc, name)
        supplied = raw.get(name, MISSING)

        if supplied is MISSING:
            fallback = default_for(model, name)
            if fallback is MISSING:
                errors.append(
                    {
                        "loc": field_loc,
                        "type": "missing",
                        "msg": "Field required",
                        "input": raw,
                    }
                )
            else:
                values[name] = fallback
            continue

        if supplied is None:
            if nullable:
                values[name] = None
            else:
                errors.append(
                    {
                        "loc": field_loc,
                        "type": f"{TYPE_NAMES.get(inner, getattr(inner, '__name__', 'value'))}_type",
                        "msg": "Input should be a value, not null",
                        "input": None,
                    }
                )
            continue

        if isinstance(inner, type) and issubclass(inner, MiniModel):
            nested, nested_errors = validate(
                inner, supplied, loc=field_loc, allow_extra=allow_extra
            )
            if nested_errors:
                errors.extend(nested_errors)
            else:
                values[name] = nested
            continue

        converted, error_type = coerce(supplied, inner)
        if error_type is None:
            values[name] = converted
        else:
            expected = TYPE_NAMES.get(inner, getattr(inner, "__name__", "value"))
            verb = "be a valid" if error_type.endswith("_parsing") else "be"
            errors.append(
                {
                    "loc": field_loc,
                    "type": error_type,
                    "msg": f"Input should {verb} {expected}",
                    "input": supplied,
                }
            )

    if not allow_extra:
        for key in raw:
            if key not in fields:
                errors.append(
                    {
                        "loc": (*loc, key),
                        "type": "extra_forbidden",
                        "msg": "Extra inputs are not permitted",
                        "input": raw[key],
                    }
                )

    if errors:
        return None, errors
    return model(**values), []


class ValidationReport:
    """A tiny wrapper so a caller can ask a report questions instead of a list."""

    def __init__(self, errors: list[dict[str, Any]]) -> None:
        self.errors = errors

    def __len__(self) -> int:
        return len(self.errors)

    def __bool__(self) -> bool:
        return bool(self.errors)

    def types(self) -> list[str]:
        return [error["type"] for error in self.errors]

    def locations(self) -> list[tuple[Any, ...]]:
        return [error["loc"] for error in self.errors]

    def at(self, *loc: Any) -> list[dict[str, Any]]:
        return [error for error in self.errors if error["loc"] == loc]


def format_report(errors: list[dict[str, Any]]) -> str:
    """Render errors the way a human wants to read them: location first."""
    if not errors:
        return "no errors"
    lines = [f"{len(errors)} validation error(s)"]
    for error in errors:
        where = ".".join(str(part) for part in error["loc"]) or "<root>"
        lines.append(f"  {where}")
        lines.append(f"    {error['msg']} [type={error['type']}, input={error['input']!r}]")
    return "\n".join(lines)

"""Going back out: ``model_dump``, ``model_dump_json``, and where the round trip breaks.

Validation is only half a boundary. Data has to leave too, and the assumption
that ``Model.model_validate(instance.model_dump())`` always works is one of the
most common things people get wrong about pydantic — and it is wrong for
ordinary, sensible reasons rather than exotic ones.

This module demonstrates each of them by doing it. Run it directly:

    python3 examples/serialize.py
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from models import Reading
from pydantic import TypeAdapter, ValidationError

LAB_DIR = Path(__file__).resolve().parent.parent
BATCH = LAB_DIR / "data" / "raw-readings.json"


def first_valid_record() -> dict[str, Any]:
    with BATCH.open(encoding="utf-8") as handle:
        return json.load(handle)[0]


def _outcome(payload: Any) -> str:
    try:
        Reading.model_validate(payload)
    except ValidationError as exc:
        return "refused: " + ", ".join(
            f"{'.'.join(str(p) for p in e['loc'])} [{e['type']}]" for e in exc.errors()
        )
    return "accepted"


def main() -> int:
    reading = Reading.model_validate(first_valid_record())

    print("1. Two dumps, two different jobs")
    print()
    dumped = reading.model_dump()
    print(f"model_dump()      -> recorded_at is a {type(dumped['recorded_at']).__name__}")
    print(f"model_dump_json() -> {reading.model_dump_json()}")
    print()
    print("model_dump gives Python objects; model_dump_json gives a JSON string and")
    print("has to turn the datetime into text on the way. Reach for the second when")
    print("the destination is a file or a socket, and the first when it is more Python.")
    print()

    print("2. The field name is not the wire name")
    print()
    print(f"default keys  : {list(dumped)}")
    print(f"by_alias=True : {list(reading.model_dump(by_alias=True))}")
    print()
    print("The model reads `pm2_5` on the way in because of the alias, and writes")
    print("`pm25` on the way out unless you ask for by_alias. If the thing at the")
    print("other end is the same vendor system, you almost certainly want by_alias.")
    print()

    print("3. The round trip is not symmetric")
    print()
    print(f"model_validate(model_dump())                  -> {_outcome(dumped)}")
    trimmed = reading.model_dump(by_alias=True, exclude={"band"})
    print(f"model_validate(model_dump(by_alias, -band))   -> {_outcome(trimmed)}")
    print()
    print("`band` is a computed_field. It is serialised because a consumer wants it,")
    print("and it is refused on the way back in because `extra='forbid'` is doing its")
    print("job: nothing computed is an input. Both behaviours are correct; the bug")
    print("would be assuming they compose.")
    print()

    print("4. Trimming the output at the point of use")
    print()
    print(f"exclude={{'operator'}}   -> {list(reading.model_dump(exclude={'operator'}))}")
    print(f"exclude_none=True     -> {list(reading.model_dump(exclude_none=True))}")
    print(f"include={{'reading_id'}} -> {reading.model_dump(include={'reading_id'})}")
    print()
    print("`operator` is a name. It is in the record because the pipeline needs")
    print("provenance, and it is excluded here because the published extract does")
    print("not. Deciding that at the serialiser rather than in six call sites is")
    print("the whole reason these arguments exist.")
    print()

    print("5. TypeAdapter: validation for things that are not models")
    print()
    batch_adapter = TypeAdapter(list[Reading])
    print("TypeAdapter(list[Reading]) validates a whole list in one call")
    good = [first_valid_record()]
    print(f"  one good record  -> {len(batch_adapter.validate_python(good))} Reading object(s)")
    try:
        batch_adapter.validate_python(good + [{"reading_id": "RD-0100"}])
    except ValidationError as exc:
        locs = [tuple(e["loc"]) for e in exc.errors()]
        print(f"  one bad appended -> {len(exc.errors())} errors, loc[0]={locs[0]}")
        print("     note the leading index: loc tells you WHICH element failed")
    print()
    ints = TypeAdapter(list[int])
    print(f"TypeAdapter(list[int]).validate_python(['1', '2']) -> {ints.validate_python(['1', '2'])}")
    print(f"TypeAdapter(list[int]).dump_json([1, 2])           -> {ints.dump_json([1, 2])!r}")
    print()

    print("6. The schema, for free")
    print()
    schema = Reading.model_json_schema()
    print(f"required fields : {sorted(schema['required'])}")
    print(f"pm2_5 property  : {json.dumps(schema['properties']['pm2_5'], sort_keys=True)}")
    print()
    print("That is a JSON Schema document, generated from the annotations. It is")
    print("what FastAPI publishes as OpenAPI, and it is what you hand a language")
    print("model when you want structured output back.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

lesson_id: D094
day: 94
kind: guided-build
languages: [python, bash]
setup_commands:
  - cd labs/sections/programming-with-python/day-094-data-validation-with-pydantic
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
  - .venv/bin/python3 -c "import pydantic; print(pydantic.VERSION)"
run_commands:
  - .venv/bin/python3 examples/coercion.py
  - .venv/bin/python3 examples/scratch_demo.py
  - .venv/bin/python3 examples/serialize.py
  - .venv/bin/python3 examples/gate.py
  - '.venv/bin/python3 examples/gate.py --fail-over 0.1  # optional: exits 1 when too much of the batch is bad'
  - .venv/bin/python3 starter/byhand.py
  - .venv/bin/pytest tests
  - .venv/bin/pytest starter
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - rm -rf out
  - rm -rf .pytest_cache tests/.pytest_cache starter/.pytest_cache
  - "find . -type d -name '__pycache__' -prune -exec rm -rf -- {} +"
  - 'rm -rf .venv  # optional: removes the lab virtual environment'
  - 'git checkout -- starter/  # optional: reset your work'
requires_network: true
requires_api_key: false
estimated_minutes: 35
last_executed: '2026-08-16'
executed_on: 'macOS 26.5.2 (Apple Silicon, arm64), Python 3.14.0, pydantic 2.13.4, pydantic-core 2.46.4, pytest 9.1.1, bash 3.2.57 — bash tests/run_tests.sh -> 62 checks, 0 failure(s), exit 0; pytest tests -> 47 passed; pytest starter -> 1 passed, 9 skipped; examples/gate.py -> read 12 records, accepted 4, rejected 8, exit 0. Network is needed once to install the two pinned packages; nothing in the lab opens a socket at run time. pydantic-settings is a separate distribution and is deliberately NOT installed here, so the lesson describes it and reproduces no output from it. This run used PYTHON/PYTEST overrides pointing at an interpreter outside the lab directory rather than a lab-local .venv; the harness resolves either.'

pydantic==2.13.4
pytest==9.1.1

"""The "before" picture: boundary validation written entirely by hand.

Nothing in this file is broken. It runs, it is correct as far as it goes, and
it is roughly what every codebase grows before somebody reaches for a library.
Read it once and count the things it does not check — that count is the point.

Run it to see it work:

    python3 starter/byhand.py

All names, station codes and operator initials in the sample data are invented
for this lab. No real monitoring network, person or measurement is involved.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

LAB_DIR = Path(__file__).resolve().parent.parent
BATCH = LAB_DIR / "data" / "raw-readings.json"


def validate_reading_by_hand(raw: Any) -> tuple[dict[str, Any] | None, list[str]]:
    """Check one record. Returns ``(clean_record, problems)``.

    Notice the shape of the code, not the detail: one ``if`` per field per rule,
    every message written out longhand, and the rules living nowhere except
    inside this function. Add a field and you edit here. Add a caller and you
    hope they remember to call this. Change a rule and you grep.
    """
    problems: list[str] = []

    if not isinstance(raw, dict):
        return None, ["record is not an object"]

    clean: dict[str, Any] = {}

    reading_id = raw.get("reading_id")
    if reading_id is None:
        problems.append("reading_id is missing")
    elif not isinstance(reading_id, str):
        problems.append("reading_id is not a string")
    else:
        clean["reading_id"] = reading_id

    station = raw.get("station")
    if station is None:
        problems.append("station is missing")
    elif not isinstance(station, dict):
        problems.append("station is not an object")
    else:
        code = station.get("code")
        if not isinstance(code, str):
            problems.append("station.code is missing or not a string")
        else:
            clean["station_code"] = code

    pm = raw.get("pm2_5")
    if pm is None:
        problems.append("pm2_5 is missing")
    else:
        try:
            clean["pm25"] = float(pm)
        except (TypeError, ValueError):
            problems.append("pm2_5 is not a number")

    humidity = raw.get("humidity_pct")
    if humidity is None:
        problems.append("humidity_pct is missing")
    else:
        try:
            clean["humidity_pct"] = int(humidity)
        except (TypeError, ValueError):
            problems.append("humidity_pct is not a whole number")

    if problems:
        return None, problems
    return clean, []


def main() -> int:
    with BATCH.open(encoding="utf-8") as handle:
        records = json.load(handle)

    accepted = 0
    rejected = 0
    for index, raw in enumerate(records):
        clean, problems = validate_reading_by_hand(raw)
        if problems:
            rejected += 1
            print(f"  record {index}: " + "; ".join(problems))
        else:
            accepted += 1

    print()
    print(f"by hand: accepted {accepted}, rejected {rejected} of {len(records)}")
    print()
    print("Four fields checked out of eight, no ranges, no patterns, no dates, no")
    print("nesting past one level, no cross-field rules, and the error messages are")
    print("prose a machine cannot act on. Every one of those is an exercise below.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""Exercises 7-10 — the data-quality gate.

The rule this file exists to enforce is simple to state and easy to get wrong:
**one bad record must not end the run.** A pipeline that raises on the first
malformed row processes nothing and tells you about one problem. A pipeline
with a gate processes everything it can and hands back a report naming every
record it refused and exactly why.

This file runs as it stands and tells you what is still missing:

    python3 starter/gate.py

The reference answer is in `examples/gate.py`.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

LAB_DIR = Path(__file__).resolve().parent.parent
DEFAULT_INPUT = LAB_DIR / "data" / "raw-readings.json"


def load_records(path: Path = DEFAULT_INPUT) -> list[Any]:
    """Read the raw batch. Note what this does *not* do: it does not validate.

    This one is written for you, because reading a file is not the lesson.
    """
    with path.open(encoding="utf-8") as handle:
        data = json.load(handle)
    if not isinstance(data, list):
        raise ValueError(f"{path} must contain a JSON array of records")
    return data


# ---------------------------------------------------------------------------
# Exercise 7 — the two result types.
#
# Define a frozen dataclass `Rejection` with fields:
#     index: int
#     reading_id: str | None
#     errors: list[dict[str, Any]]
# and a `summary()` method returning a one-line human string such as
#     "record 4 (RD-0005): humidity_pct [less_than_equal]"
#
# Define a dataclass `GateResult` with `accepted: list` and
# `rejected: list[Rejection]` (use field(default_factory=list)), plus
# properties `seen`, `accepted_count`, `rejected_count`, a method
# `error_types()` returning every error type across all rejections, and
# `as_report()` returning a JSON-serialisable dict with keys
# records_seen / records_accepted / records_rejected / rejections.
#
# Keep `loc`, `type`, `msg` and `input` for each error. Three of those four are
# machine-readable; `msg` is the one for humans, and the one nothing asserts on.
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Exercise 8 — run_gate(records) -> GateResult.
#
# For each record, in order:
#     try:  reading = Reading.model_validate(raw)
#     except ValidationError as exc:  record a Rejection built from
#         exc.errors(), then `continue` — do NOT re-raise.
#
# The `except` is the entire exercise. Everything else is bookkeeping. If you
# find yourself letting the exception out "just for now", stop: that is the
# behaviour this file exists to prevent.
#
# Tip: pull the raw id with raw.get("reading_id") BEFORE validating, so a
# record that fails validation can still be named in the report.
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Exercise 9 — the batch-level rule the schema cannot express.
#
# Inside run_gate, keep a dict of {reading_id: first_index_seen}. If a record
# validates but its id has already been used, reject it with a hand-made error
# entry of type "duplicate_id" and loc ["reading_id"].
#
# Uniqueness is a property of the BATCH, not of the record. No per-record
# schema can see it, which is why a gate is a place and not just a model.
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Exercise 10 — write the two outputs.
#
# write_outputs(result, out_dir) should create out_dir if needed and write:
#   * accepted.jsonl — one `reading.model_dump_json()` per line
#   * rejects.json   — json.dumps(result.as_report(), indent=2)
# returning the two paths.
#
# Then make main() print the counts and one line per rejection, and support a
# `--fail-over FRACTION` flag that exits 1 when the reject rate is too high.
# A gate that never fails the build is a log line, not a gate.
# ---------------------------------------------------------------------------


def _unfinished() -> list[str]:
    return [
        name
        for name in ("Rejection", "GateResult", "run_gate", "write_outputs")
        if name not in globals()
    ]


def main() -> int:
    missing = _unfinished()
    records = load_records()
    print(f"loaded {len(records)} raw records from {DEFAULT_INPUT.name}")
    if missing:
        print(f"gate not built yet: {', '.join(missing)} not defined.")
        print("Start at exercise 7.")
        return 0
    result = run_gate(records)  # noqa: F821 - defined by exercise 8
    print(f"accepted {result.accepted_count}, rejected {result.rejected_count}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""Exercises 1-6 — replace `byhand.py` with a schema.

This file runs as it stands. It just does not do anything yet, and it says so
when you run it:

    python3 starter/models.py

Work through the numbered exercises in order. After each one, run:

    PYTEST=/path/to/pytest  # or use the lab's .venv
    "$PYTEST" starter -q

and delete the matching `@pytest.mark.skip` line in `starter/test_starter.py`
as each exercise starts passing. The reference answer for every exercise is in
`examples/models.py`; open it after you have tried, not before.

All names, station codes and operator initials in the sample data are invented
for this lab. No real monitoring network, person or measurement is involved.
"""

from __future__ import annotations

# ---------------------------------------------------------------------------
# Exercise 1 — the imports and the two Annotated types.
#
#   from datetime import datetime
#   from typing import Annotated
#   from pydantic import (
#       BaseModel, ConfigDict, Field, StringConstraints,
#       computed_field, field_validator, model_validator,
#   )
#
# Then declare the constrained types ONCE, so no field has to repeat them:
#
#   StationCode = Annotated[str, StringConstraints(pattern=r"^ST-[A-Z]{3}$")]
#   ReadingId   = Annotated[str, StringConstraints(pattern=r"^RD-\d{4}$")]
#   Percent     = Annotated[int, Field(ge=0, le=100)]
#   Micrograms  = Annotated[float, Field(ge=0.0, le=1000.0)]
#
# Check it: python3 -c "from pydantic import TypeAdapter; ..."  or just move on
# to exercise 2 — the tests will tell you.
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Exercise 2 — the nested model.
#
# Define `class Station(BaseModel)` with:
#   * model_config = ConfigDict(extra="forbid", str_strip_whitespace=True,
#                               frozen=True)
#   * code: StationCode
#   * name: str, at least 1 and at most 60 characters (Field(min_length=...))
#   * elevation_m: int between -500 and 9000
#
# `extra="forbid"` is the one that catches a misspelled key. Without it, a
# record with `humidty_pct` validates happily and loses the field in silence.
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Exercise 3 — the record model, and three words that are not synonyms.
#
# Define `class Reading(BaseModel)` with model_config = ConfigDict(
#     extra="forbid", str_strip_whitespace=True,
#     validate_assignment=True, populate_by_name=True)
#
# Fields:
#   reading_id: ReadingId
#   station: Station                      <- a nested model, validated in turn
#   recorded_at: datetime
#   pm25: Micrograms = Field(alias="pm2_5")   <- the vendor's name on the wire
#   temperature_c: float between -90.0 and 60.0
#   humidity_pct: Percent
#   operator: str | None                  <- REQUIRED and NULLABLE: the key must
#                                            be there; its value may be null
#   notes: str | None = None              <- OPTIONAL and nullable: it has a
#                                            default, so the key may be absent
#
# Required, optional and nullable are three separate facts. `operator` and
# `notes` differ only in that one has a default, and that difference is the
# whole distinction. Prove it to yourself: after this exercise,
# `Reading.model_json_schema()["required"]` should list `operator` and not
# `notes`.
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Exercise 4 — two field validators.
#
# Add to Reading, each decorated @field_validator("<name>", mode="after") and
# @classmethod underneath it (order matters — field_validator goes on top):
#
#   * blank_note_is_no_note("notes"): if the value is a string that strips to
#     empty, return None instead. An empty string is not a note.
#   * timestamp_must_carry_a_timezone("recorded_at"): if value.tzinfo is None,
#     raise ValueError(...). A naive timestamp is an ambiguity, not a time.
#
# Raise plain ValueError inside a validator. pydantic catches it and folds it
# into the ValidationError with type "value_error" — you never raise
# ValidationError yourself.
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Exercise 5 — the cross-field rule.
#
# Add @model_validator(mode="after") named a_high_reading_must_be_explained.
# It takes `self` (mode="after" runs on the built object) and returns `self`.
#
# The rule: if self.pm25 > 500.0 and there is no non-blank self.notes, raise
# ValueError. Above 500 the instrument is either witnessing something serious
# or misbehaving, and the difference is not in the number — so the schema
# demands a human wrote down which it was.
#
# No single-field constraint can express this, because it depends on two
# fields at once. That is exactly when you reach for model_validator.
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Exercise 6 — a computed field.
#
# Add a `band` property decorated with @computed_field on top of @property,
# returning "good" (<=12.0), "moderate" (<=35.4),
# "unhealthy-for-sensitive-groups" (<=55.4), "unhealthy" (<=150.4), else
# "hazardous", based on self.pm25.
#
# Then run examples/serialize.py and notice what it does to the round trip.
# ---------------------------------------------------------------------------


def _unfinished() -> str:
    missing = [name for name in ("Station", "Reading") if name not in globals()]
    if not missing:
        return "Station and Reading are defined. Run the starter tests."
    return f"No schema yet: {', '.join(missing)} not defined. Start at exercise 1."


if __name__ == "__main__":
    print(_unfinished())

[pytest]
# The starter modules import each other by bare name (`from models import
# Reading`), so this directory has to be on sys.path. `rootdir`-relative
# insertion is what pytest does by default for a directory with no packages,
# but stating it here means the suite behaves the same however it is invoked.
pythonpath = .
addopts = -p no:cacheprovider
testpaths = .

"""The starter suite. One test passes today; nine are waiting for you.

Delete the `@pytest.mark.skip(...)` line above a test as soon as the matching
exercise is done, then run the suite again:

    "$PYTEST" starter -q

Every assertion below is on an error's `type` or its `loc`. Not one of them
looks at `msg`. That is deliberate and it is the habit worth taking away from
today: `type` and `loc` are the machine-readable contract, and `msg` is prose
the library is free to reword in any release. The same argument applied to a
FastAPI 422 body on Day 082; it is the same body.

All names, station codes and operator initials here are invented for this lab.
"""

from __future__ import annotations

import json
from pathlib import Path

import pytest

LAB_DIR = Path(__file__).resolve().parent.parent
BATCH = LAB_DIR / "data" / "raw-readings.json"


def load_batch() -> list[dict]:
    with BATCH.open(encoding="utf-8") as handle:
        return json.load(handle)


def locs_and_types(exc_info) -> set[tuple[tuple, str]]:
    """Every problem as (loc, type). The only two things worth asserting on."""
    return {(tuple(e["loc"]), e["type"]) for e in exc_info.value.errors()}


VALID = {
    "reading_id": "RD-0042",
    "station": {"code": "ST-KLM", "name": "Kalmar Ridge", "elevation_m": 340},
    "recorded_at": "2026-08-15T06:00:00Z",
    "pm2_5": 12.4,
    "temperature_c": 18.2,
    "humidity_pct": 61,
    "operator": "R. Nayar",
    "notes": None,
}


# ---------------------------------------------------------------------------
# This one passes right now. It is the "before" picture, tested.
# ---------------------------------------------------------------------------


def test_the_hand_written_validator_catches_a_bad_number_and_misses_a_bad_range():
    from byhand import validate_reading_by_hand

    records = load_batch()

    # Record 3 has pm2_5 = "not-measured". float() refuses it, so the hand
    # written check does too.
    _, problems = validate_reading_by_hand(records[3])
    assert problems, "the hand-written validator should reject a non-numeric pm2_5"

    # Record 4 has humidity_pct = 118. There is no range rule anywhere in
    # byhand.py, so it sails through. This assertion documents a hole.
    clean, problems = validate_reading_by_hand(records[4])
    assert problems == []
    assert clean["humidity_pct"] == 118


# ---------------------------------------------------------------------------
# Exercise 2
# ---------------------------------------------------------------------------


@pytest.mark.skip(reason="Exercise 2: define Station in starter/models.py")
def test_station_rejects_a_code_that_does_not_match_the_pattern():
    from models import Station
    from pydantic import ValidationError

    with pytest.raises(ValidationError) as exc_info:
        Station(code="ST-north", name="Northgate Yard", elevation_m=210)
    assert (("code",), "string_pattern_mismatch") in locs_and_types(exc_info)


@pytest.mark.skip(reason="Exercise 2: set extra='forbid' on Station")
def test_station_refuses_an_unexpected_key():
    from models import Station
    from pydantic import ValidationError

    with pytest.raises(ValidationError) as exc_info:
        Station(code="ST-KLM", name="Kalmar Ridge", elevation_m=340, elevaton_m=340)
    assert (("elevaton_m",), "extra_forbidden") in locs_and_types(exc_info)


# ---------------------------------------------------------------------------
# Exercise 3
# ---------------------------------------------------------------------------


@pytest.mark.skip(reason="Exercise 3: required, optional and nullable are three things")
def test_required_optional_and_nullable_are_three_different_things():
    from models import Reading
    from pydantic import ValidationError

    required = set(Reading.model_json_schema()["required"])
    # operator is REQUIRED (the key must be present) and NULLABLE (may be null).
    assert "operator" in required
    # notes is OPTIONAL (it has a default) and also nullable.
    assert "notes" not in required

    # Nullable: an explicit null is fine.
    assert Reading.model_validate({**VALID, "operator": None}).operator is None

    # Required: leaving the key out is not.
    without_operator = {k: v for k, v in VALID.items() if k != "operator"}
    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate(without_operator)
    assert (("operator",), "missing") in locs_and_types(exc_info)

    # Optional: leaving `notes` out is fine, and the default arrives.
    without_notes = {k: v for k, v in VALID.items() if k != "notes"}
    assert Reading.model_validate(without_notes).notes is None


@pytest.mark.skip(reason="Exercise 3: give pm25 the alias 'pm2_5'")
def test_the_alias_is_the_name_on_the_wire_and_the_name_in_the_error():
    from models import Reading
    from pydantic import ValidationError

    reading = Reading.model_validate(VALID)
    assert reading.pm25 == pytest.approx(12.4)

    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate({**VALID, "pm2_5": "not-measured"})
    # The error names the key the caller actually sent, which is what makes the
    # report usable by whoever owns the source file.
    assert (("pm2_5",), "float_parsing") in locs_and_types(exc_info)


@pytest.mark.skip(reason="Exercise 3: lax mode coerces, strict mode does not")
def test_lax_mode_coerces_numeric_strings_and_strict_mode_refuses_them():
    from models import Reading
    from pydantic import ValidationError

    stringy = {
        **VALID,
        "station": {"code": "ST-KLM", "name": "  Kalmar Ridge  ", "elevation_m": "340"},
        "pm2_5": "14.8",
        "temperature_c": "19",
        "humidity_pct": "58",
    }

    lax = Reading.model_validate(stringy)
    assert lax.station.elevation_m == 340
    assert lax.pm25 == pytest.approx(14.8)
    assert lax.humidity_pct == 58
    # str_strip_whitespace is on, so the padded name arrives trimmed.
    assert lax.station.name == "Kalmar Ridge"

    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate(stringy, strict=True)
    found = locs_and_types(exc_info)
    assert (("pm2_5",), "float_type") in found
    assert (("humidity_pct",), "int_type") in found


# ---------------------------------------------------------------------------
# Exercise 4 and 5
# ---------------------------------------------------------------------------


@pytest.mark.skip(reason="Exercise 4: constrain humidity_pct to 0-100")
def test_an_out_of_range_percentage_is_refused_with_a_range_error_type():
    from models import Reading
    from pydantic import ValidationError

    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate({**VALID, "humidity_pct": 118})
    assert (("humidity_pct",), "less_than_equal") in locs_and_types(exc_info)


@pytest.mark.skip(reason="Exercise 5: the cross-field model_validator")
def test_a_high_reading_needs_a_note_and_the_error_sits_on_the_whole_record():
    from models import Reading
    from pydantic import ValidationError

    hot = {**VALID, "pm2_5": 612.5, "notes": None}
    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate(hot)
    # loc is empty: no single field is at fault, the combination is.
    assert ((), "value_error") in locs_and_types(exc_info)

    explained = {**hot, "notes": "smoke plume from the north, confirmed by the duty log"}
    assert Reading.model_validate(explained).pm25 == pytest.approx(612.5)


@pytest.mark.skip(reason="Exercises 2-5: a ValidationError reports everything at once")
def test_one_call_reports_every_problem_rather_than_the_first():
    from models import Reading
    from pydantic import ValidationError

    broken = {
        "reading_id": "RD-0099",
        "station": {"code": "ST-QQQ", "name": "Quarry Head", "elevation_m": "high"},
        "recorded_at": "2026-08-15T13:00:00Z",
        "pm2_5": "unreadable",
        "temperature_c": 20.0,
        "humidity_pct": None,
        "operator": "A. Invented",
    }
    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate(broken)

    found = locs_and_types(exc_info)
    assert (("station", "elevation_m"), "int_parsing") in found
    assert (("pm2_5",), "float_parsing") in found
    assert (("humidity_pct",), "int_type") in found
    # Three problems, one exception, one round trip to fix them all.
    assert len(exc_info.value.errors()) == 3


# ---------------------------------------------------------------------------
# Exercises 7-10
# ---------------------------------------------------------------------------


@pytest.mark.skip(reason="Exercises 7-9: build the gate in starter/gate.py")
def test_the_gate_finishes_the_whole_batch_with_a_non_zero_reject_count():
    import gate

    records = gate.load_records(BATCH)
    # The whole point: this call must RETURN, not raise, on a batch that is
    # two-thirds bad. If a ValidationError escapes here, the gate is not a gate.
    result = gate.run_gate(records)

    assert result.seen == len(records)
    assert result.rejected_count > 0
    assert result.accepted_count + result.rejected_count == result.seen

    types = set(result.error_types())
    for expected in (
        "missing",
        "float_parsing",
        "less_than_equal",
        "extra_forbidden",
        "string_pattern_mismatch",
        "duplicate_id",
        "datetime_from_date_parsing",
        "value_error",
    ):
        assert expected in types, f"the batch should have produced a {expected}"

    # Every rejection can name itself well enough to fix the source.
    for rejection in result.rejected:
        assert rejection.errors
        for error in rejection.errors:
            assert isinstance(error["type"], str) and error["type"]
            assert isinstance(list(error["loc"]), list)

[pytest]
addopts = -p no:cacheprovider
testpaths = .

#!/usr/bin/env bash
# Tests for the Day 094 lab. Run from the lab directory:
#   bash tests/run_tests.sh
#
# This harness proves the claims the lesson makes, and proves them by running
# the code rather than by reading it:
#
#   * the schema refuses each of the eight planted problems, and the refusal
#     carries a machine-readable `type` and `loc`;
#   * lax mode really performs the conversions the lesson tabulates, and
#     strict mode really refuses them;
#   * the gate completes a two-thirds-bad batch with a non-zero reject count
#     instead of raising — the single most important behaviour in the lab;
#   * the miniature from-scratch validator collects every error rather than
#     the first, and lets through exactly the records it has no rules for;
#   * the starter suite is not vacuous: it goes fully green against the
#     reference implementation, and it goes RED when one rule is removed.
#
# Deterministic, non-interactive, offline. Exits 0 only if every check passes.
set -u

export PYTHONDONTWRITEBYTECODE=1

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

# Bytecode left by an EARLIER command is not this run's litter. The README
# documents `pytest starter -q`, and running it writes .pyc files that would
# then fail the cleanliness check at the end of this script -- failing the
# reader for following the instructions. Clearing them here makes that final
# check measure what it claims to: what THIS run left behind. `.venv` is
# untouched, because the packages' own bytecode is theirs, not ours.
find "${lab_dir}" -name '.venv' -prune -o -type d -name '__pycache__' -exec rm -rf {} + 2>/dev/null || true
find "${lab_dir}" -name '.venv' -prune -o -type d -name '.pytest_cache' -exec rm -rf {} + 2>/dev/null || true

failures=0
checks=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 the tools: an explicit override, then this lab's own virtual
# environment, then whatever is on PATH. Fails loudly with instructions rather
# than skipping silently — a suite that quietly does nothing is worse than one
# that stops.
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
}

install_hint() {
  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 environment:" >&2
  echo "    PYTHON=/path/to/python3 PYTEST=/path/to/pytest bash tests/run_tests.sh" >&2
}

pytest_bin="$(resolve_tool pytest "${PYTEST:-}")" || {
  echo "FAIL: pytest not found." >&2
  install_hint
  exit 1
}

# The Python that owns that pytest is the one with pydantic installed, unless
# an explicit PYTHON says otherwise.
if [ -n "${PYTHON:-}" ] && [ -x "${PYTHON}" ]; then
  python_bin="${PYTHON}"
else
  python_bin="$(dirname "${pytest_bin}")/python3"
  [ -x "${python_bin}" ] || python_bin="$(command -v python3 || true)"
fi
if [ -z "${python_bin}" ] || [ ! -x "${python_bin}" ]; then
  echo "FAIL: python3 not found." >&2
  install_hint
  exit 1
fi

if ! "${python_bin}" -c "import pydantic" >/dev/null 2>&1; then
  echo "FAIL: pydantic is not importable from ${python_bin}." >&2
  install_hint
  exit 1
fi

echo "Day 094 — Guard the Boundary"
echo

# --------------------------------------------------------------------------
echo "1. The tools and the versions this lab was written against"
# --------------------------------------------------------------------------

echo "     python  $("${python_bin}" -c 'import sys; print(sys.version.split()[0])')"
versions="$("${python_bin}" - <<'PY'
from importlib.metadata import version
for name in ("pydantic", "pydantic_core", "pytest"):
    try:
        print(f"{name}=={version(name)}")
    except Exception:
        print(f"{name}==<not installed>")
PY
)"
printf '%s\n' "${versions}" | sed 's/^/     /'

for pin in "pydantic==2.13.4" "pytest==9.1.1"; do
  case "${versions}" in
    *"${pin}"*) check "installed ${pin} matches requirements/requirements.txt" "yes" ;;
    *) check "installed ${pin} matches requirements/requirements.txt" "no" ;;
  esac
done

# pydantic-settings is a separate distribution and this lab does not install
# it. The lesson describes it and reproduces no output from it; this check
# keeps that claim honest by confirming it really is absent here.
if "${python_bin}" -c "import pydantic_settings" >/dev/null 2>&1; then
  check "pydantic-settings is absent, as the lesson states" "no"
else
  check "pydantic-settings is absent, as the lesson states" "yes"
fi

# The v2 API surface the lesson teaches is the one that actually exists.
api_out="$("${python_bin}" - <<'PY' 2>&1
import pydantic

v2 = [
    "BaseModel", "Field", "ConfigDict", "TypeAdapter",
    "field_validator", "model_validator", "computed_field",
    "StringConstraints", "ValidationError",
]
missing = [name for name in v2 if not hasattr(pydantic, name)]
print("missing:" + ",".join(missing) if missing else "all-present")
print("model_validate:" + str(hasattr(pydantic.BaseModel, "model_validate")))
print("model_dump:" + str(hasattr(pydantic.BaseModel, "model_dump")))
PY
)"
case "${api_out}" in
  *all-present*) check "every pydantic v2 name the lesson uses exists" "yes" ;;
  *) check "every pydantic v2 name the lesson uses exists (${api_out})" "no" ;;
esac
case "${api_out}" in
  *"model_validate:True"*"model_dump:True"*)
    check "BaseModel exposes model_validate and model_dump (v2, not v1)" "yes" ;;
  *) check "BaseModel exposes model_validate and model_dump (v2, not v1)" "no" ;;
esac

# --------------------------------------------------------------------------
echo
echo "2. The reference suite passes"
# --------------------------------------------------------------------------

tests_out="$(cd "${lab_dir}" && "${pytest_bin}" tests -q 2>&1)"
tests_exit=$?
if [ "${tests_exit}" -eq 0 ]; then
  check "pytest tests exits 0" "yes"
else
  check "pytest tests exits 0 (got ${tests_exit})" "no"
  printf '%s\n' "${tests_out}" | tail -40
fi
case "${tests_out}" in
  *"47 passed"*) check "pytest tests reports 47 passed" "yes" ;;
  *) check "pytest tests reports 47 passed (got: $(printf '%s' "${tests_out}" | tail -1))" "no" ;;
esac

collected="$(cd "${lab_dir}" && "${pytest_bin}" tests --collect-only -q 2>&1)"
for test_id in \
  "test_required_optional_and_nullable_are_three_different_things" \
  "test_one_call_reports_every_problem_at_once" \
  "test_every_error_entry_carries_loc_type_msg_and_input" \
  "test_strict_mode_refuses_the_same_strings" \
  "test_the_round_trip_is_not_symmetric_and_here_is_exactly_why" \
  "test_the_miniature_validator_collects_every_error_rather_than_the_first" \
  "test_the_gate_completes_the_batch_with_a_non_zero_reject_count" \
  "test_every_problem_the_brief_planted_is_actually_caught"
do
  case "${collected}" in
    *"${test_id}"*) check "collection finds ${test_id}" "yes" ;;
    *) check "collection finds ${test_id}" "no" ;;
  esac
done

# No test in this lab may assert on an error's prose. `type` and `loc` are the
# stable interface; `msg` is not. This is a grep, and it is deliberate.
if grep -nE "\[.msg.\]|errors\(\)\[0\]\[.msg.\]" \
     "${lab_dir}/tests/test_validation.py" "${lab_dir}/starter/test_starter.py" \
     | grep -vE "^\S+:[0-9]+: *#" | grep -q "assert"; then
  check "no test asserts on an error message string" "no"
else
  check "no test asserts on an error message string" "yes"
fi

# --------------------------------------------------------------------------
echo
echo "3. The gate runs the whole batch and reports what it refused"
# --------------------------------------------------------------------------

gate_dir="$(mktemp -d "${TMPDIR:-/tmp}/day094-gate.XXXXXX")"
gate_out="$(cd "${lab_dir}/examples" && "${python_bin}" gate.py \
  --input "${lab_dir}/data/raw-readings.json" --out-dir "${gate_dir}" 2>&1)"
gate_exit=$?
printf '%s\n' "${gate_out}" | sed 's/^/     /'
if [ "${gate_exit}" -eq 0 ]; then
  check "examples/gate.py exits 0 on a batch that is two-thirds bad" "yes"
else
  check "examples/gate.py exits 0 on a batch that is two-thirds bad (got ${gate_exit})" "no"
fi

for fragment in \
  'read      12 records' \
  'accepted  4' \
  'rejected  8' \
  'record 2 (RD-0003): operator [missing]' \
  'record 3 (RD-0004): pm2_5 [float_parsing]' \
  'record 4 (RD-0005): humidity_pct [less_than_equal]' \
  'humidty_pct [extra_forbidden]' \
  'record 6 (RD-0007): station.code [string_pattern_mismatch]' \
  'record 7 (RD-0001): reading_id [duplicate_id]' \
  'record 8 (RD-0009): recorded_at [datetime_from_date_parsing]' \
  'record 9 (RD-0010): <record> [value_error]'
do
  case "${gate_out}" in
    *"${fragment}"*) check "gate output contains: ${fragment}" "yes" ;;
    *) check "gate output contains: ${fragment}" "no" ;;
  esac
done

accepted_lines="$(wc -l < "${gate_dir}/accepted.jsonl" | tr -d ' ')"
if [ "${accepted_lines}" = "4" ]; then
  check "accepted.jsonl holds 4 records" "yes"
else
  check "accepted.jsonl holds 4 records (counted ${accepted_lines})" "no"
fi

report_check="$("${python_bin}" - "${gate_dir}/rejects.json" <<'PY' 2>&1
import json, sys
report = json.load(open(sys.argv[1], encoding="utf-8"))
assert report["records_seen"] == 12, report["records_seen"]
assert report["records_accepted"] == 4
assert report["records_rejected"] == 8
assert len(report["rejections"]) == 8
for rejection in report["rejections"]:
    for error in rejection["errors"]:
        assert set(error) >= {"loc", "type", "msg", "input"}, sorted(error)
print("report-ok")
PY
)"
case "${report_check}" in
  *report-ok*) check "rejects.json names all 8 refusals with loc/type/msg/input" "yes" ;;
  *) check "rejects.json names all 8 refusals with loc/type/msg/input (${report_check})" "no" ;;
esac

# The threshold flag really fails the build.
(cd "${lab_dir}/examples" && "${python_bin}" gate.py \
  --input "${lab_dir}/data/raw-readings.json" --out-dir "${gate_dir}" \
  --fail-over 0.1 >/dev/null 2>&1)
if [ $? -ne 0 ]; then
  check "--fail-over 0.1 exits non-zero on a 67% reject rate" "yes"
else
  check "--fail-over 0.1 exits non-zero on a 67% reject rate" "no"
fi
rm -rf "${gate_dir}"

# --------------------------------------------------------------------------
echo
echo "4. The demo scripts run and print what the lesson quotes"
# --------------------------------------------------------------------------

coercion_out="$(cd "${lab_dir}/examples" && "${python_bin}" coercion.py 2>&1)"
coercion_exit=$?
if [ "${coercion_exit}" -eq 0 ]; then
  check "examples/coercion.py exits 0" "yes"
else
  check "examples/coercion.py exits 0 (got ${coercion_exit})" "no"
fi
# Each of these is a claim the lesson makes about default coercion, checked
# against the line the program actually printed.
for fragment in \
  "'42'                     int          42                                 refused: int_type" \
  "'forty-two'              int          refused: int_parsing" \
  "42.7                     int          refused: int_from_float" \
  "True                     int          1  " \
  "42                       str          refused: string_type" \
  "3                        float        3.0                                3.0"
do
  case "${coercion_out}" in
    *"${fragment}"*) check "coercion table shows: $(echo "${fragment}" | tr -s ' ')" "yes" ;;
    *) check "coercion table shows: $(echo "${fragment}" | tr -s ' ')" "no" ;;
  esac
done

scratch_out="$(cd "${lab_dir}/examples" && "${python_bin}" scratch_demo.py 2>&1)"
scratch_exit=$?
if [ "${scratch_exit}" -eq 0 ]; then
  check "examples/scratch_demo.py exits 0" "yes"
else
  check "examples/scratch_demo.py exits 0 (got ${scratch_exit})" "no"
fi
for fragment in \
  'from scratch : accepted 9, rejected 3' \
  'pydantic     : accepted 5, rejected 7' \
  'record 4: pydantic says less_than_equal; the toy has no rule for it' \
  'record 8: pydantic says datetime_from_date_parsing; the toy has no rule for it'
do
  case "${scratch_out}" in
    *"${fragment}"*) check "scratch_demo shows: ${fragment}" "yes" ;;
    *) check "scratch_demo shows: ${fragment}" "no" ;;
  esac
done

serialize_out="$(cd "${lab_dir}/examples" && "${python_bin}" serialize.py 2>&1)"
serialize_exit=$?
if [ "${serialize_exit}" -eq 0 ]; then
  check "examples/serialize.py exits 0" "yes"
else
  check "examples/serialize.py exits 0 (got ${serialize_exit})" "no"
fi
for fragment in \
  'model_validate(model_dump())                  -> refused: band [extra_forbidden]' \
  'model_validate(model_dump(by_alias, -band))   -> accepted' \
  "required fields : ['humidity_pct', 'operator', 'pm2_5', 'reading_id', 'recorded_at', 'station', 'temperature_c']" \
  'loc[0]=(1, '"'"'station'"'"')'
do
  case "${serialize_out}" in
    *"${fragment}"*) check "serialize shows: ${fragment}" "yes" ;;
    *) check "serialize shows: ${fragment}" "no" ;;
  esac
done

# `notes` is optional, so it is absent from `required`; `operator` is required
# and nullable, so it is present. The line above asserts both at once.
case "${serialize_out}" in
  *"required fields : ['humidity_pct', 'operator',"*"'notes'"*)
    check "notes is absent from the required list" "no" ;;
  *) check "notes is absent from the required list" "yes" ;;
esac

# --------------------------------------------------------------------------
echo
echo "5. The starter is runnable before you start, and honest about it"
# --------------------------------------------------------------------------

starter_out="$(cd "${lab_dir}" && "${pytest_bin}" starter -q 2>&1)"
starter_exit=$?
if [ "${starter_exit}" -eq 0 ]; then
  check "pytest starter exits 0 with the exercises unfinished" "yes"
else
  check "pytest starter exits 0 with the exercises unfinished (got ${starter_exit})" "no"
fi
case "${starter_out}" in
  *"1 passed, 9 skipped"*) check "the starter has 1 worked test and 9 skipped exercises" "yes" ;;
  *) check "the starter has 1 worked test and 9 skipped exercises" "no" ;;
esac

byhand_out="$(cd "${lab_dir}" && "${python_bin}" starter/byhand.py 2>&1)"
case "${byhand_out}" in
  *"by hand: accepted 10, rejected 2 of 12"*)
    check "starter/byhand.py runs and shows the hand-written validator's blind spots" "yes" ;;
  *) check "starter/byhand.py runs and shows the hand-written validator's blind spots" "no" ;;
esac

models_out="$(cd "${lab_dir}" && "${python_bin}" starter/models.py 2>&1)"
case "${models_out}" in
  *"No schema yet: Station, Reading not defined"*)
    check "starter/models.py reports its unfinished state honestly" "yes" ;;
  *) check "starter/models.py reports its unfinished state honestly" "no" ;;
esac

gate_starter_out="$(cd "${lab_dir}" && "${python_bin}" starter/gate.py 2>&1)"
case "${gate_starter_out}" in
  *"gate not built yet"*)
    check "starter/gate.py reports its unfinished state honestly" "yes" ;;
  *) check "starter/gate.py reports its unfinished state honestly" "no" ;;
esac

# --------------------------------------------------------------------------
echo
echo "6. The starter suite is not vacuous — green when solved, red when broken"
# --------------------------------------------------------------------------

# Drop the reference implementation in as the student's answer, un-skip
# everything, and demand a fully green run. A suite that cannot tell a finished
# schema from an unfinished one is worth nothing.
work="$(mktemp -d "${TMPDIR:-/tmp}/day094-solved.XXXXXX")"
mkdir -p "${work}/lab"
cp -R "${lab_dir}/data" "${work}/data"
cp "${lab_dir}/starter/pytest.ini" "${lab_dir}/starter/byhand.py" "${work}/lab/"
cp "${lab_dir}/examples/models.py" "${lab_dir}/examples/gate.py" "${work}/lab/"
grep -v '^@pytest\.mark\.skip' "${lab_dir}/starter/test_starter.py" > "${work}/lab/test_starter.py"

solved_out="$(cd "${work}/lab" && "${pytest_bin}" . -q 2>&1)"
solved_exit=$?
if [ "${solved_exit}" -eq 0 ]; then
  check "the starter suite goes fully green against the finished schema" "yes"
else
  check "the starter suite goes fully green against the finished schema (exit ${solved_exit})" "no"
  printf '%s\n' "${solved_out}" | tail -20
fi
case "${solved_out}" in
  *"10 passed"*) check "all 10 starter tests pass once the exercises are done" "yes" ;;
  *) check "all 10 starter tests pass once the exercises are done" "no" ;;
esac

# Now break exactly one rule — widen the percentage constraint from 0-100 to
# 0-1000 — and demand the suite FAILS. This is the check that proves the range
# assertion is doing work rather than passing by accident.
"${python_bin}" - "${work}/lab/models.py" <<'PY'
import sys
from pathlib import Path

path = Path(sys.argv[1])
text = path.read_text(encoding="utf-8")
broken = text.replace("Annotated[int, Field(ge=0, le=100)]", "Annotated[int, Field(ge=0, le=1000)]")
assert broken != text, "the Percent constraint was not found — this check would be vacuous"
path.write_text(broken, encoding="utf-8")
PY
broken_out="$(cd "${work}/lab" && "${pytest_bin}" . -q 2>&1)"
broken_exit=$?
if [ "${broken_exit}" -ne 0 ]; then
  check "widening the percentage range makes the suite FAIL (exit ${broken_exit}, not 0)" "yes"
else
  check "widening the percentage range makes the suite FAIL — it did not, so the range check is vacuous" "no"
fi
case "${broken_out}" in
  *"test_an_out_of_range_percentage_is_refused_with_a_range_error_type"*)
    check "the failing run names the range check by test id" "yes" ;;
  *) check "the failing run names the range check by test id" "no" ;;
esac
rm -rf "${work}"

# --------------------------------------------------------------------------
echo
echo "7. The lab left nothing behind"
# --------------------------------------------------------------------------

# `.venv` is deliberately NOT in this list. The README tells the reader to
# create it, and the tool resolution at the top of this file looks inside
# it — so treating it as litter would fail the lab for following its own
# setup instructions.
for stray in "out"; do
  if [ -e "${lab_dir}/${stray}" ]; then
    check "no ${stray}/ left inside the lab after a full run" "no"
  else
    check "no ${stray}/ left inside the lab after a full run" "yes"
  fi
done

# `.venv` is pruned from the searches below. A virtual environment ships the
# installed packages' own precompiled bytecode -- hundreds of __pycache__
# directories that came with NumPy or pytest and have nothing to do with
# whether THIS lab tidied up after itself. Without the prune, following the
# README's own setup instructions makes this check fail, which reports a
# problem the reader cannot fix and did not cause.
if find "${lab_dir}" -name '.venv' -prune -o -type d -name '__pycache__' -print -quit 2>/dev/null | grep -q .; then
  check "no __pycache__ left inside the lab after a full run" "no"
else
  check "no __pycache__ left inside the lab after a full run" "yes"
fi

# Nothing here reaches the network at run time. The only network step is the
# one-off pip install described in the README.
# Restricted to .py files on purpose: this script quotes the pattern it is
# searching for, so scanning itself would always match.
if find "${lab_dir}/examples" "${lab_dir}/starter" "${lab_dir}/tests" -name '*.py' -print0 2>/dev/null \
     | xargs -0 grep -qE 'requests\.|urlopen|httpx\.|socket\.(create_connection|socket)\(' 2>/dev/null; then
  check "no lab source opens a network connection at run time" "no"
else
  check "no lab source opens a network connection at run time" "yes"
fi

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

"""The reference suite for Day 094 — the schema, the gate, and the toy.

Every assertion here is on an error's `type` or its `loc`, on a count, or on a
validated value. Not one of them reads `msg`. `type` and `loc` are the parts of
a `ValidationError` pydantic treats as an interface; `msg` is prose it is free
to reword between releases, exactly as a FastAPI 422 body's `msg` is (Day 082).
A suite that greps error text passes until the day somebody improves a sentence.

Run from the lab directory:

    "$PYTEST" tests -q

All names, station codes and operator initials in the fixtures are invented for
this lab. No real monitoring network, person or measurement is involved.
"""

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path

import pytest

LAB_DIR = Path(__file__).resolve().parent.parent
EXAMPLES = LAB_DIR / "examples"
BATCH = LAB_DIR / "data" / "raw-readings.json"

sys.path.insert(0, str(EXAMPLES))

from gate import GateResult, load_records, run_gate, write_outputs  # noqa: E402
from models import Reading, Station  # noqa: E402
from pydantic import TypeAdapter, ValidationError  # noqa: E402
from scratch_models import ScratchReading  # noqa: E402
from scratch_validator import validate as mini_validate  # noqa: E402

VALID = {
    "reading_id": "RD-0042",
    "station": {"code": "ST-KLM", "name": "Kalmar Ridge", "elevation_m": 340},
    "recorded_at": "2026-08-15T06:00:00Z",
    "pm2_5": 12.4,
    "temperature_c": 18.2,
    "humidity_pct": 61,
    "operator": "R. Nayar",
    "notes": None,
}


def problems(exc_info) -> set[tuple[tuple, str]]:
    """Every error as (loc, type) — the two stable parts."""
    return {(tuple(e["loc"]), e["type"]) for e in exc_info.value.errors()}


@pytest.fixture(scope="module")
def records() -> list[dict]:
    return load_records(BATCH)


@pytest.fixture(scope="module")
def result(records) -> GateResult:
    return run_gate(records)


# ---------------------------------------------------------------------------
# The schema
# ---------------------------------------------------------------------------


def test_a_clean_record_validates_into_a_typed_object():
    reading = Reading.model_validate(VALID)
    assert reading.reading_id == "RD-0042"
    assert isinstance(reading.station, Station)
    assert reading.recorded_at.tzinfo is not None
    assert reading.pm25 == pytest.approx(12.4)
    assert reading.band == "moderate"


def test_required_optional_and_nullable_are_three_different_things():
    required = set(Reading.model_json_schema()["required"])
    assert "operator" in required, "operator is required — the key must be present"
    assert "notes" not in required, "notes is optional — it has a default"

    assert Reading.model_validate({**VALID, "operator": None}).operator is None

    without_operator = {k: v for k, v in VALID.items() if k != "operator"}
    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate(without_operator)
    assert (("operator",), "missing") in problems(exc_info)

    without_notes = {k: v for k, v in VALID.items() if k != "notes"}
    assert Reading.model_validate(without_notes).notes is None


def test_a_misspelled_key_is_caught_rather_than_silently_dropped():
    typo = {k: v for k, v in VALID.items() if k != "humidity_pct"}
    typo["humidty_pct"] = 61
    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate(typo)
    found = problems(exc_info)
    assert (("humidty_pct",), "extra_forbidden") in found
    assert (("humidity_pct",), "missing") in found


def test_the_nested_model_reports_its_own_location():
    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate(
            {**VALID, "station": {"code": "ST-north", "name": "Northgate Yard", "elevation_m": 210}}
        )
    assert (("station", "code"), "string_pattern_mismatch") in problems(exc_info)


def test_an_out_of_range_value_is_refused_by_the_annotated_constraint():
    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate({**VALID, "humidity_pct": 118})
    assert (("humidity_pct",), "less_than_equal") in problems(exc_info)


def test_a_date_in_the_wrong_format_is_refused():
    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate({**VALID, "recorded_at": "15/08/2026 10:00"})
    assert (("recorded_at",), "datetime_from_date_parsing") in problems(exc_info)


def test_a_naive_timestamp_is_refused_by_the_field_validator():
    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate({**VALID, "recorded_at": "2026-08-15T06:00:00"})
    # A validator that raises ValueError surfaces as value_error at that field.
    assert (("recorded_at",), "value_error") in problems(exc_info)


def test_the_cross_field_rule_puts_its_error_on_the_whole_record():
    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate({**VALID, "pm2_5": 612.5, "notes": None})
    assert ((), "value_error") in problems(exc_info)

    explained = {
        **VALID,
        "pm2_5": 548.9,
        "notes": "smoke plume from the north, confirmed by the duty log",
    }
    assert Reading.model_validate(explained).band == "hazardous"


def test_a_blank_note_is_normalised_to_none():
    assert Reading.model_validate({**VALID, "notes": "   "}).notes is None
    assert Reading.model_validate({**VALID, "notes": "  routine sweep  "}).notes == "routine sweep"


def test_one_call_reports_every_problem_at_once():
    broken = {
        "reading_id": "RD-0099",
        "station": {"code": "ST-QQQ", "name": "Quarry Head", "elevation_m": "high"},
        "recorded_at": "2026-08-15T13:00:00Z",
        "pm2_5": "unreadable",
        "temperature_c": 20.0,
        "humidity_pct": None,
        "operator": "A. Invented",
    }
    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate(broken)
    found = problems(exc_info)
    assert (("station", "elevation_m"), "int_parsing") in found
    assert (("pm2_5",), "float_parsing") in found
    assert (("humidity_pct",), "int_type") in found
    assert len(exc_info.value.errors()) == 3


def test_every_error_entry_carries_loc_type_msg_and_input():
    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate({**VALID, "humidity_pct": 118})
    entry = exc_info.value.errors()[0]
    assert set(entry) >= {"loc", "type", "msg", "input"}
    assert entry["input"] == 118, "input echoes back what you actually sent"


# ---------------------------------------------------------------------------
# Coercion: lax versus strict
# ---------------------------------------------------------------------------


def test_lax_mode_coerces_numeric_strings():
    stringy = {
        **VALID,
        "station": {"code": "ST-KLM", "name": "  Kalmar Ridge  ", "elevation_m": "340"},
        "pm2_5": "14.8",
        "temperature_c": "19",
        "humidity_pct": "58",
    }
    reading = Reading.model_validate(stringy)
    assert reading.station.elevation_m == 340
    assert reading.station.name == "Kalmar Ridge", "str_strip_whitespace trims on the way in"
    assert reading.pm25 == pytest.approx(14.8)
    assert reading.temperature_c == pytest.approx(19.0)
    assert reading.humidity_pct == 58


def test_strict_mode_refuses_the_same_strings():
    stringy = {**VALID, "pm2_5": "14.8", "humidity_pct": "58"}
    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate(stringy, strict=True)
    found = problems(exc_info)
    assert (("pm2_5",), "float_type") in found
    assert (("humidity_pct",), "int_type") in found


@pytest.mark.parametrize(
    ("value", "target", "expected"),
    [
        ("42", int, 42),
        ("42.0", int, 42),
        ("  42  ", int, 42),
        (42.0, int, 42),
        (True, int, 1),
        (3, float, 3.0),
        ("true", bool, True),
        ((1, 2), list[int], [1, 2]),
    ],
)
def test_the_conversions_lax_mode_really_performs(value, target, expected):
    assert TypeAdapter(target).validate_python(value) == expected


@pytest.mark.parametrize(
    ("value", "target", "error_type"),
    [
        ("forty-two", int, "int_parsing"),
        (42.7, int, "int_from_float"),
        (42, str, "string_type"),
        ("[1, 2]", list[int], "list_type"),
    ],
)
def test_the_conversions_lax_mode_refuses(value, target, error_type):
    with pytest.raises(ValidationError) as exc_info:
        TypeAdapter(target).validate_python(value)
    assert exc_info.value.errors()[0]["type"] == error_type


def test_int_to_float_is_the_one_conversion_strict_mode_still_allows():
    assert TypeAdapter(float).validate_python(3, strict=True) == pytest.approx(3.0)
    with pytest.raises(ValidationError):
        TypeAdapter(int).validate_python("42", strict=True)


# ---------------------------------------------------------------------------
# Assignment, immutability, serialization
# ---------------------------------------------------------------------------


def test_validate_assignment_keeps_the_object_legal_after_construction():
    reading = Reading.model_validate(VALID)
    with pytest.raises(ValidationError) as exc_info:
        reading.humidity_pct = 500
    assert (("humidity_pct",), "less_than_equal") in problems(exc_info)
    assert reading.humidity_pct == 61, "the rejected assignment did not land"


def test_a_frozen_model_refuses_assignment_outright():
    station = Reading.model_validate(VALID).station
    with pytest.raises(ValidationError) as exc_info:
        station.elevation_m = 1
    # Observed in pydantic 2.13.4: a whole-model `frozen=True` reports
    # `frozen_instance`, not the per-field `frozen_field`. The distinction is
    # exactly the sort of thing that makes asserting on `msg` a losing game.
    assert (("elevation_m",), "frozen_instance") in problems(exc_info)


def test_the_round_trip_is_not_symmetric_and_here_is_exactly_why():
    reading = Reading.model_validate(VALID)
    dumped = reading.model_dump()
    assert "band" in dumped, "a computed field is serialised"

    with pytest.raises(ValidationError) as exc_info:
        Reading.model_validate(dumped)
    assert (("band",), "extra_forbidden") in problems(exc_info)

    trimmed = reading.model_dump(by_alias=True, exclude={"band"})
    assert Reading.model_validate(trimmed) == reading


def test_the_alias_decides_the_key_on_both_sides():
    reading = Reading.model_validate(VALID)
    assert "pm25" in reading.model_dump()
    assert "pm2_5" in reading.model_dump(by_alias=True)
    assert "pm2_5" in json.loads(reading.model_dump_json(by_alias=True))


def test_model_dump_json_turns_the_datetime_into_iso_8601_text():
    reading = Reading.model_validate(VALID)
    assert json.loads(reading.model_dump_json())["recorded_at"] == "2026-08-15T06:00:00Z"


def test_type_adapter_validates_a_list_and_names_the_failing_index():
    adapter = TypeAdapter(list[Reading])
    assert len(adapter.validate_python([VALID])) == 1
    with pytest.raises(ValidationError) as exc_info:
        adapter.validate_python([VALID, {"reading_id": "RD-0100"}])
    first_loc = tuple(exc_info.value.errors()[0]["loc"])
    assert first_loc[0] == 1, "loc leads with the index of the offending element"


# ---------------------------------------------------------------------------
# The from-scratch validator
# ---------------------------------------------------------------------------


def test_the_miniature_validator_collects_every_error_rather_than_the_first():
    broken = {
        "reading_id": "RD-0099",
        "station": {"code": "ST-QQQ", "name": "Quarry Head", "elevation_m": "high"},
        "recorded_at": "2026-08-15T13:00:00Z",
        "pm2_5": "unreadable",
        "temperature_c": 20.0,
        "humidity_pct": None,
        "operator": "A. Invented",
    }
    value, errors = mini_validate(ScratchReading, broken)
    assert value is None
    found = {(tuple(e["loc"]), e["type"]) for e in errors}
    assert (("station", "elevation_m"), "int_parsing") in found
    assert (("pm2_5",), "float_parsing") in found
    assert (("humidity_pct",), "int_type") in found
    assert len(errors) == 3


def test_the_miniature_validator_distinguishes_absent_from_null():
    base = {
        "reading_id": "RD-0042",
        "station": {"code": "ST-KLM", "name": "Kalmar Ridge", "elevation_m": 340},
        "recorded_at": "2026-08-15T06:00:00Z",
        "pm2_5": 12.4,
        "temperature_c": 18.2,
        "humidity_pct": 61,
        "operator": "R. Nayar",
    }
    # `operator` present and null: legal, it is nullable.
    value, errors = mini_validate(ScratchReading, {**base, "operator": None})
    assert errors == [] and value is not None and value.operator is None

    # `operator` absent: not legal, it has no default.
    missing = {k: v for k, v in base.items() if k != "operator"}
    _, errors = mini_validate(ScratchReading, missing)
    assert (("operator",), "missing") in {(tuple(e["loc"]), e["type"]) for e in errors}


def test_the_miniature_validator_refuses_a_bool_where_an_int_was_asked_for():
    # bool is a subclass of int in Python. A validator that forgets this lets a
    # checkbox arrive where a count was wanted.
    base = {
        "reading_id": "RD-0042",
        "station": {"code": "ST-KLM", "name": "Kalmar Ridge", "elevation_m": True},
        "recorded_at": "2026-08-15T06:00:00Z",
        "pm2_5": 12.4,
        "temperature_c": 18.2,
        "humidity_pct": 61,
        "operator": "R. Nayar",
    }
    _, errors = mini_validate(ScratchReading, base)
    assert (("station", "elevation_m"), "int_type") in {
        (tuple(e["loc"]), e["type"]) for e in errors
    }


def test_the_toy_accepts_records_pydantic_refuses_because_it_has_no_such_rules(records):
    # Record 4 is humidity 118, record 6 is a malformed station code. The toy
    # checks types and presence and nothing else, so both pass it.
    for index in (4, 6):
        value, errors = mini_validate(ScratchReading, records[index])
        assert errors == [] and value is not None
        with pytest.raises(ValidationError):
            Reading.model_validate(records[index])


# ---------------------------------------------------------------------------
# The gate
# ---------------------------------------------------------------------------


def test_the_gate_completes_the_batch_with_a_non_zero_reject_count(records, result):
    # This is the assertion the whole lab exists for: run_gate RETURNS on a
    # batch that is two-thirds bad. If a ValidationError escaped, this test
    # would error rather than fail, and the gate would not be a gate.
    assert result.seen == len(records) == 12
    assert result.rejected_count == 8
    assert result.accepted_count == 4
    assert result.accepted_count + result.rejected_count == result.seen


def test_every_problem_the_brief_planted_is_actually_caught(result):
    types = set(result.error_types())
    for expected in (
        "missing",  # a required field absent
        "float_parsing",  # a value that is genuinely not a number
        "less_than_equal",  # an out-of-range value
        "extra_forbidden",  # a misspelled field name
        "string_pattern_mismatch",  # a nested object with its own error
        "duplicate_id",  # a batch rule no schema can express
        "datetime_from_date_parsing",  # a date in the wrong format
        "value_error",  # the cross-field rule
    ):
        assert expected in types, f"expected a {expected} somewhere in the batch"


def test_a_number_arriving_as_a_string_is_accepted_by_coercion(result):
    accepted = {reading.reading_id: reading for reading in result.accepted}
    assert "RD-0002" in accepted, "the all-strings record should be coerced, not rejected"
    coerced = accepted["RD-0002"]
    assert coerced.pm25 == pytest.approx(14.8)
    assert coerced.humidity_pct == 58
    assert coerced.station.elevation_m == 340


def test_the_duplicate_is_the_second_occurrence_not_the_first(result):
    accepted_ids = [reading.reading_id for reading in result.accepted]
    assert accepted_ids.count("RD-0001") == 1, "the first RD-0001 was kept"
    duplicates = [
        rejection
        for rejection in result.rejected
        if any(error["type"] == "duplicate_id" for error in rejection.errors)
    ]
    assert len(duplicates) == 1
    assert duplicates[0].index == 7, "the later record is the one rejected"


def test_every_rejection_names_a_location_and_a_type(result):
    for rejection in result.rejected:
        assert rejection.errors, "a rejection with no reasons is useless"
        for error in rejection.errors:
            assert isinstance(error["type"], str) and error["type"]
            assert isinstance(error["loc"], list)


def test_the_report_is_json_serialisable_and_counts_add_up(result):
    report = result.as_report()
    text = json.dumps(report)  # must not raise: `input` was made JSON-friendly
    assert json.loads(text) == report
    assert report["records_seen"] == report["records_accepted"] + report["records_rejected"]
    assert len(report["rejections"]) == report["records_rejected"]


def test_write_outputs_emits_the_accepted_records_and_the_report(tmp_path, result):
    accepted_path, report_path = write_outputs(result, tmp_path)
    lines = accepted_path.read_text(encoding="utf-8").strip().splitlines()
    assert len(lines) == result.accepted_count
    assert json.loads(lines[0])["reading_id"] == "RD-0001"
    report = json.loads(report_path.read_text(encoding="utf-8"))
    assert report["records_rejected"] == result.rejected_count


def test_the_gate_can_fail_the_build_when_the_reject_rate_is_too_high(tmp_path):
    import gate

    ok = gate.main(
        ["--input", str(BATCH), "--out-dir", str(tmp_path / "a"), "--fail-over", "0.9"]
    )
    assert ok == 0, "a 67% reject rate is under a 90% threshold"
    bad = gate.main(
        ["--input", str(BATCH), "--out-dir", str(tmp_path / "b"), "--fail-over", "0.1"]
    )
    assert bad == 1, "a 67% reject rate is over a 10% threshold"


# ---------------------------------------------------------------------------
# The scripts run
# ---------------------------------------------------------------------------


@pytest.mark.parametrize("script", ["coercion.py", "scratch_demo.py", "serialize.py"])
def test_each_demo_script_exits_zero(script):
    done = subprocess.run(
        [sys.executable, script],
        cwd=EXAMPLES,
        capture_output=True,
        text=True,
        env={"PYTHONDONTWRITEBYTECODE": "1", "PATH": "/usr/bin:/bin"},
    )
    assert done.returncode == 0, done.stderr
    assert done.stdout.strip(), "a demo that prints nothing teaches nothing"