Hands-on lab — Day 97: Logging and Configuration

#!/usr/bin/env python3
"""The script we are starting from. Everything it says, it says with print().

Run it:

    python3 examples/01_prints.py

Then read the output and ask the four questions that matter once this script
is running somewhere you are not:

    1. WHEN did each line happen? There is no timestamp. If two of these
       lines are three hours apart you cannot tell.
    2. WHICH RUN is this? If it runs hourly, yesterday's output and today's
       are identical text in the same file with nothing to separate them.
    3. HOW BAD is each line? "skipping record" and "could not write the
       output" are the same shape of text. `grep` cannot tell them apart, so
       neither can an alert.
    4. HOW DO I TURN IT DOWN? You edit the file, and you deploy. There is no
       other way, because the decision to print is baked into every call.

And one more, which is the one that gets people fired: line 3 of the output
prints the API key. It is now in the terminal scrollback, in the CI job log,
and in whatever file the output was redirected to.

Nothing here is a straw man. This is what a working script looks like before
anybody has needed to operate it.
"""

# A tiny data-preparation job. Six records in, some of them bad.
RECORDS = [
    {"id": 1, "text": "the cat sat on the mat", "label": "neutral"},
    {"id": 2, "text": "", "label": "neutral"},
    {"id": 3, "text": "shipping was late and the box was crushed", "label": "negative"},
    {"id": 4, "text": "arrived early, works perfectly", "label": "positive"},
    {"id": 5, "text": "no opinion", "label": "unknown"},
    {"id": 6, "text": "great value for the price", "label": "positive"},
]

VALID_LABELS = {"neutral", "negative", "positive"}
API_KEY = "sk-live-9f2c4a7b1e63"  # invented for this lab; not a real credential


def prepare(records):
    print("starting preparation")
    print("using API key " + API_KEY)  # never do this. It is done here on purpose.
    kept = []
    for record in records:
        print("processing record " + str(record["id"]))
        if not record["text"]:
            print("skipping record " + str(record["id"]) + ": empty text")
            continue
        if record["label"] not in VALID_LABELS:
            print("skipping record " + str(record["id"]) + ": unknown label")
            continue
        kept.append(record)
    print("kept " + str(len(kept)) + " of " + str(len(records)) + " records")
    return kept


def main():
    kept = prepare(RECORDS)
    # The actual RESULT of the program, mixed into the same stream as all the
    # commentary above. A caller that wants to pipe the result somewhere gets
    # the commentary too.
    for record in kept:
        print(f"{record['id']}\t{record['label']}\t{record['text']}")
    print("done")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""The logging module's architecture, demonstrated one confusing behaviour at a time.

    python3 examples/02_logging_architecture.py

Six demonstrations, each one a thing that surprises people:

    A. the same script as 01_prints.py, converted
    B. the two-level trap — a logger at DEBUG whose handler is at WARNING,
       and the message that vanishes
    C. propagation, the duplicate message it causes, and two fixes
    D. exception() against error(str(e)), and what the second one throws away
    E. lazy formatting, and what it actually saves
    F. the five levels, chosen honestly

Every demonstration captures its own log output into a StringIO buffer rather
than letting it go to stdout, then prints the buffer. That is not a testing
trick bolted on afterwards — it is how you should capture logs in a test, and
it is why the output below is exactly the records that reached each handler
and nothing else.

One line of sanitising: demonstration D prints a real traceback, and a
traceback contains the absolute path of this file. The script rewrites that
path to `<lab>` before printing, so the captured output in expected-output/ is
identical on every machine. Nothing else is altered.
"""

from __future__ import annotations

import io
import logging
from pathlib import Path

LAB_DIR = Path(__file__).resolve().parent.parent


def banner(letter: str, title: str) -> None:
    print()
    print("=" * 70)
    print(f"{letter}. {title}")
    print("=" * 70)


def fresh_logger(name: str) -> logging.Logger:
    """A logger with no handlers and nothing inherited, for a clean demo.

    Reaching into `logger.handlers` like this is fine in a demonstration and
    is not how you configure a real application — `dictConfig` is, and
    demonstration 05 shows it. It is done here so each section starts from a
    known state regardless of what the section before it did.
    """
    logger = logging.getLogger(name)
    logger.handlers.clear()
    logger.filters.clear()
    logger.setLevel(logging.NOTSET)
    logger.propagate = True
    return logger


def attach_buffer(
    logger: logging.Logger, level: int, fmt: str = "%(levelname)-8s %(name)-16s %(message)s"
) -> io.StringIO:
    stream = io.StringIO()
    handler = logging.StreamHandler(stream)
    handler.setLevel(level)
    handler.setFormatter(logging.Formatter(fmt))
    logger.addHandler(handler)
    return stream


def show(stream: io.StringIO, label: str) -> None:
    text = stream.getvalue()
    print(f"--- {label} ---")
    if not text.strip():
        print("(nothing. Not one line.)")
    else:
        print(text.rstrip())


# ---------------------------------------------------------------------------
# A. The converted script
# ---------------------------------------------------------------------------

RECORDS = [
    {"id": 1, "text": "the cat sat on the mat", "label": "neutral"},
    {"id": 2, "text": "", "label": "neutral"},
    {"id": 3, "text": "shipping was late and the box was crushed", "label": "negative"},
    {"id": 4, "text": "arrived early, works perfectly", "label": "positive"},
    {"id": 5, "text": "no opinion", "label": "unknown"},
    {"id": 6, "text": "great value for the price", "label": "positive"},
]
VALID_LABELS = {"neutral", "negative", "positive"}


def prepare(records, log: logging.Logger):
    """01_prints.py, with every print replaced by the level it deserved.

    Note what changed besides the function name. Each line now carries a
    severity, so an operator can ask for INFO and above and never see the
    per-record chatter. The result of the function is RETURNED rather than
    printed, so the log and the output are two different streams. And the
    API key does not appear at all, because it was never the log's business.
    """
    log.info("preparation starting: %d records", len(records))
    kept = []
    for record in records:
        log.debug("processing record %s", record["id"])
        if not record["text"]:
            log.warning("skipping record %s: empty text", record["id"])
            continue
        if record["label"] not in VALID_LABELS:
            log.warning("skipping record %s: unknown label %r", record["id"], record["label"])
            continue
        kept.append(record)
    log.info("preparation done: kept %d of %d", len(kept), len(records))
    return kept


def demo_a() -> None:
    banner("A", "The same job, converted: severity, and a result you can pipe")
    log = fresh_logger("prep")
    log.setLevel(logging.DEBUG)
    log.propagate = False

    everything = attach_buffer(log, logging.DEBUG)
    prepare(RECORDS, log)
    show(everything, "the developer's view: handler at DEBUG")

    log = fresh_logger("prep")
    log.setLevel(logging.DEBUG)
    log.propagate = False
    operator = attach_buffer(log, logging.INFO)
    kept = prepare(RECORDS, log)
    show(operator, "the operator's view: SAME CODE, handler at INFO")

    print()
    print("The two views came from one unchanged function. That is the whole")
    print("difference from print(): the decision about what is worth seeing")
    print("moved out of the call site and into configuration.")
    print(f"(the function returned {len(kept)} records, which never touched the log)")


# ---------------------------------------------------------------------------
# B. The two-level trap
# ---------------------------------------------------------------------------


def demo_b() -> None:
    banner("B", "The two-level trap: the message that vanishes")
    log = fresh_logger("trap")
    log.setLevel(logging.DEBUG)          # the logger will accept DEBUG
    log.propagate = False
    stream = attach_buffer(log, logging.WARNING)  # the handler will not emit it

    log.debug("this debug line is accepted by the logger and dropped by the handler")
    log.info("so is this info line")
    log.warning("this warning gets through")

    print("logger level:  DEBUG   (logging.getLogger('trap').level ->",
          logging.getLevelName(log.level) + ")")
    print("handler level: WARNING (log.handlers[0].level ->",
          logging.getLevelName(log.handlers[0].level) + ")")
    show(stream, "three calls were made; this is what came out")
    print()
    print("A record has to pass TWO level checks, and they belong to two")
    print("different objects. The logger's check happens first and decides")
    print("whether a LogRecord is created at all. Each handler then applies")
    print("its own. Setting the logger to DEBUG and wondering where your")
    print("debug output went is the single most common logging question")
    print("there is, and this is the whole answer.")

    log.handlers.clear()
    stream2 = attach_buffer(log, logging.DEBUG)
    log.debug("and now, with a handler that accepts DEBUG")
    show(stream2, "the fix: lower the HANDLER's level too")


# ---------------------------------------------------------------------------
# C. Propagation and the duplicate message
# ---------------------------------------------------------------------------


def _propagation_setup(propagate: bool, app_handler: bool):
    """Build root + myapp + myapp.loader from scratch, and return the buffers."""
    root = logging.getLogger()
    root.handlers.clear()
    root.setLevel(logging.DEBUG)
    root_stream = attach_buffer(root, logging.DEBUG, "ROOT     | %(name)s | %(message)s")

    app = fresh_logger("myapp")
    app.setLevel(logging.DEBUG)
    app.propagate = propagate
    app_stream = io.StringIO()
    if app_handler:
        app_stream = attach_buffer(app, logging.DEBUG, "MYAPP    | %(name)s | %(message)s")

    child = logging.getLogger("myapp.loader")
    child.handlers.clear()
    child.filters.clear()
    child.setLevel(logging.NOTSET)
    child.propagate = True
    return child, app_stream, root_stream


def demo_c() -> None:
    banner("C", "Propagation: why you are seeing everything twice")

    root = logging.getLogger()
    saved_handlers, saved_level = root.handlers[:], root.level

    child, app_stream, root_stream = _propagation_setup(propagate=True, app_handler=True)
    child.info("loaded 3 files")

    print("Handlers: one on the root logger, one on 'myapp'. One call, made")
    print("on 'myapp.loader'.")
    show(app_stream, "what the myapp handler wrote")
    show(root_stream, "what the root handler wrote")
    print()
    print("Two lines for one call. A record travels UP the dotted hierarchy —")
    print("myapp.loader, then myapp, then root — and every handler it passes")
    print("emits it. The ancestors' LEVELS are not consulted on the way up,")
    print("only their handlers. That is deliberate and it is why the duplicate")
    print("surprises people.")
    print()
    print("The usual cause is logging.basicConfig(), which quietly puts a")
    print("handler on the ROOT logger. Add one of your own and you now have")
    print("two.")

    # Fix 1: stop the record travelling any further up than myapp.
    child, app_stream, root_stream = _propagation_setup(propagate=False, app_handler=True)
    child.info("loaded 3 files")
    show(app_stream, "fix 1: myapp.propagate = False — myapp handler")
    show(root_stream, "fix 1: myapp.propagate = False — root handler")

    # Fix 2: the better one for an application — configure ONE place.
    child, app_stream, root_stream = _propagation_setup(propagate=True, app_handler=False)
    child.info("loaded 3 files")
    show(app_stream, "fix 2: handlers in ONE place only — myapp handler")
    show(root_stream, "fix 2: handlers in ONE place only — root handler")
    print()
    print("Both fixes work and they are not equivalent. propagate = False is")
    print("the right answer for a LIBRARY that must not have its records")
    print("escape into an application it knows nothing about. 'configure one")
    print("place' is the right answer for an APPLICATION, because the")
    print("alternative is a tree of loggers each with an opinion about where")
    print("its output goes, and no single place to change it.")

    root.handlers.clear()
    root.handlers.extend(saved_handlers)
    root.setLevel(saved_level)


# ---------------------------------------------------------------------------
# D. exception() against error(str(e))
# ---------------------------------------------------------------------------


def parse_batch_size(text: str) -> int:
    return int(text)


def demo_d() -> None:
    banner("D", "exception() against error(str(e)): the traceback is the value")

    log = fresh_logger("parse")
    log.setLevel(logging.DEBUG)
    log.propagate = False
    poor = attach_buffer(log, logging.DEBUG, "%(levelname)-8s %(message)s")

    try:
        parse_batch_size("sixty-four")
    except ValueError as error:
        log.error("could not parse batch size: %s", str(error))
    show(poor, "log.error(str(e)) — what the on-call engineer receives")

    log.handlers.clear()
    good = attach_buffer(log, logging.DEBUG, "%(levelname)-8s %(message)s")
    try:
        parse_batch_size("sixty-four")
    except ValueError:
        # Inside an except block, exception() is error() plus exc_info=True.
        # It reads the exception currently being handled out of the
        # interpreter, so you do not pass it anything.
        log.exception("could not parse batch size")
    text = good.getvalue().replace(str(LAB_DIR), "<lab>")
    print("--- log.exception() — the same failure ---")
    print(text.rstrip())
    print()
    print("The first version says a value was bad. The second says WHICH LINE")
    print("of WHICH FUNCTION was called with it, and by whom. On a machine you")
    print("cannot attach a debugger to, that difference is the difference")
    print("between a fix and a guess.")
    print()
    print("str(e) also loses the exception's TYPE. 'invalid literal for int()")
    print("with base 10' happens to name it; plenty of exceptions have empty")
    print("messages and str(e) then logs an empty string.")


# ---------------------------------------------------------------------------
# E. Lazy formatting
# ---------------------------------------------------------------------------


class ExpensiveToRender:
    """Counts how many times something asked for its string form."""

    renders = 0

    def __str__(self) -> str:
        ExpensiveToRender.renders += 1
        return "a summary that cost real work to produce"


def demo_e() -> None:
    banner("E", "Lazy formatting: log.info('saw %s', x) rather than an f-string")

    log = fresh_logger("lazy")
    log.setLevel(logging.INFO)          # DEBUG will be rejected by the logger
    log.propagate = False
    attach_buffer(log, logging.INFO)

    ExpensiveToRender.renders = 0
    for _ in range(1000):
        log.debug("summary: %s", ExpensiveToRender())     # lazy
    lazy_renders = ExpensiveToRender.renders

    ExpensiveToRender.renders = 0
    for _ in range(1000):
        log.debug(f"summary: {ExpensiveToRender()}")      # eager
    eager_renders = ExpensiveToRender.renders

    print(f"1000 suppressed DEBUG calls with %s formatting: {lazy_renders} renders")
    print(f"1000 suppressed DEBUG calls with an f-string:   {eager_renders} renders")
    print()
    print("The f-string is evaluated BEFORE logging is called, because that is")
    print("what an argument is. The %s form hands logging the object and the")
    print("template separately, and logging only joins them if a handler is")
    print("actually going to emit the record.")
    print()
    print("What it saves is therefore exactly this: the cost of rendering")
    print("arguments for records nobody wanted. On a hot path with expensive")
    print("reprs — a dataframe, a model, a large dict — that is the whole cost")
    print("of the logging call. On a cheap int it is nearly nothing, and the")
    print("honest reason to use %s everywhere is consistency plus one more")
    print("thing: the unformatted template survives onto the record as")
    print("record.msg, so a structured backend can group every occurrence of")
    print("'summary: %s' as one event with different arguments.")


# ---------------------------------------------------------------------------
# F. The five levels
# ---------------------------------------------------------------------------


def demo_f() -> None:
    banner("F", "The five levels, chosen honestly")
    log = fresh_logger("levels")
    log.setLevel(logging.DEBUG)
    log.propagate = False
    stream = attach_buffer(log, logging.DEBUG, "%(levelname)-8s (%(levelno)3d)  %(message)s")

    log.debug("retrieved 128 rows from the cache in one query")
    log.info("run 4711 started: model=small-encoder data=2026-08-01 seed=7")
    log.warning("upstream returned 429; retrying in 2s (attempt 2 of 5)")
    log.error("could not write the output file; this batch produced nothing")
    log.critical("no disk space remains; shutting down")
    show(stream, "one line at each level, with its numeric value")
    print()
    print("The question that decides the level is not 'how alarming does this")
    print("feel' but 'who is this for, and what do they do about it':")
    print()
    print("  DEBUG    for the developer, tracing their own code. Off in")
    print("           production, on when you are hunting something.")
    print("  INFO     for the operator: the run started, the run finished,")
    print("           this many records. Normal life, worth recording.")
    print("  WARNING  surprising but survivable. The retry worked. The")
    print("           deprecated flag was used. Nobody has to get up.")
    print("  ERROR    work that did not happen. This batch produced nothing.")
    print("           Somebody has to look, though not necessarily now.")
    print("  CRITICAL the process is going down and will stop doing its job.")
    print()
    print("The failure mode to avoid is level inflation. If routine events are")
    print("logged as WARNING, the warnings stop being read, and the real one")
    print("is invisible in the noise. A level is a promise to the reader about")
    print("what it costs them to ignore the line.")


def main() -> None:
    demo_a()
    demo_b()
    demo_c()
    demo_d()
    demo_e()
    demo_f()
    print()
    print("=" * 70)
    print("Six demonstrations complete.")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Structured logging as JSON, and a redacting filter that is actually tested.

    python3 examples/03_structured_logging.py

Two ideas, and the second one is the security lesson of the day.

**A log you cannot parse is a log nobody will query.** Free-text log lines are
readable by one person looking at one file. The moment there are ten thousand
of them across four machines, the question stops being "read this" and becomes
"count the ERRORs from run 4711 grouped by event", and that is a query. A query
needs fields. One JSON object per line gives you fields for the price of a
formatter subclass.

**A secret in a log line is a real incident.** Not a lint failure — an
incident, with a rotation, a disclosure conversation, and an audit of
everywhere that log was copied to. This script builds a filter that removes
known secret values, then proves the secret is absent from the captured output
rather than asserting that it should be.

Both pieces live in `examples/applog.py` so they can be imported. This file
demonstrates and explains them.
"""

from __future__ import annotations

import json
import logging
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from applog import JsonFormatter, RedactingFilter, buffer_handler  # noqa: E402

LAB_DIR = Path(__file__).resolve().parent.parent
RUN_ID = "run-4711"
API_KEY = "sk-live-9f2c4a7b1e63"  # invented for this lab; not a real credential


def sanitize(text: str) -> str:
    """Rewrite this lab's absolute path to <lab> before PRINTING captured logs.

    A rendered traceback contains the absolute path of the file it came from,
    which is different on every machine. Only the printing is affected: the
    assertions in tests/run_tests.sh run against the unmodified text.
    """
    return text.replace(str(LAB_DIR), "<lab>")


def banner(title: str) -> None:
    print()
    print("=" * 70)
    print(title)
    print("=" * 70)


def build_logger(name: str, secrets: list[str] | None = None, on_logger: bool = False):
    """A logger with one JSON handler, and optionally a redacting filter.

    `on_logger` chooses where the filter is attached — on the handler (the
    correct answer) or on the logger (the answer that looks correct and has a
    hole in it). Demonstration 2 uses both, to show the difference.
    """
    logger = logging.getLogger(name)
    logger.handlers.clear()
    logger.filters.clear()
    logger.setLevel(logging.DEBUG)
    logger.propagate = False

    handler, stream = buffer_handler(
        logging.DEBUG, JsonFormatter(static_fields={"run_id": RUN_ID})
    )
    if secrets:
        if on_logger:
            logger.addFilter(RedactingFilter(secrets))
        else:
            handler.addFilter(RedactingFilter(secrets))
    logger.addHandler(handler)
    return logger, stream


def demo_json() -> None:
    banner("1. One JSON object per line, with fields you can query")

    log, stream = build_logger("json.demo")
    log.info("run started", extra={"model": "small-encoder", "seed": 7})
    log.info("batch complete", extra={"batch": 1, "records": 64, "kept": 61})
    log.info("batch complete", extra={"batch": 2, "records": 64, "kept": 64})
    log.warning("upstream slow", extra={"attempt": 2, "status": 429})
    try:
        int("sixty-four")
    except ValueError:
        log.exception("could not parse batch size", extra={"batch": 3})
    log.info("run finished", extra={"kept_total": 125})

    lines = [line for line in stream.getvalue().splitlines() if line.strip()]
    print(f"{len(lines)} lines of JSON. Here are the first two, pretty-printed")
    print("for reading only — the real output is one object per line:")
    for line in lines[:2]:
        print(json.dumps(json.loads(line), indent=2))

    print()
    print("Now the point. That is a queryable table, so query it — with the")
    print("standard library, no log platform involved:")
    records = [json.loads(line) for line in lines]

    kept = sum(r.get("kept", 0) for r in records if r["event"] == "batch complete")
    print(f"  records kept across all batches: {kept}")

    by_level: dict[str, int] = {}
    for record in records:
        by_level[record["level"]] = by_level.get(record["level"], 0) + 1
    print(f"  lines by level: {by_level}")

    failures = [r for r in records if r["level"] == "ERROR"]
    print(f"  the failure: event={failures[0]['event']!r} "
          f"exc_type={failures[0]['exc_type']!r} batch={failures[0]['batch']}")
    print(f"  every line carries run_id={records[0]['run_id']!r}, "
          f"so this run separates from every other run")

    print()
    print("Three things earned their place there. `event` is the unformatted")
    print("message, so 'batch complete' groups across every batch. `run_id` is")
    print("a static field on the formatter, so it is stamped on every line")
    print("without any call site remembering it. And the traceback is its own")
    print("field rather than being glued onto the message, so a parser does")
    print("not have to guess where one ends and the other begins.")

    print()
    print("What it costs, stated honestly: JSON logs are unpleasant to read")
    print("with your eyes. The usual answer is JSON to the file or the")
    print("collector and a human-readable formatter on the console — the same")
    print("records, two handlers, two formatters. Demonstration 05 does that.")


def demo_redaction() -> None:
    banner("2. The redacting filter, and the proof that it worked")

    print("First, without the filter. This is what a careless line does:")
    log, stream = build_logger("redact.without")
    log.info("calling upstream with key %s", API_KEY)
    leaked = stream.getvalue().strip()
    print(f"  {leaked}")
    print(f"  the secret appears in that line: {API_KEY in leaked}")

    print()
    print("Now with the filter attached to the logger, and four different")
    print("routes by which a secret tries to get out:")
    log, stream = build_logger("redact.with", secrets=[API_KEY])

    log.info("calling upstream with key %s", API_KEY)          # via an argument
    log.info(f"hard-coded into the message: {API_KEY}")        # via the message
    log.info("config loaded", extra={"api_key": API_KEY})      # via extra=
    log.info("headers built", extra={"headers": {"Authorization": f"Bearer {API_KEY}"}})
    try:
        raise RuntimeError(f"upstream rejected key {API_KEY}")
    except RuntimeError:
        log.exception("request failed")                        # via an exception

    text = stream.getvalue()
    for line in sanitize(text).splitlines():
        print(f"  {line}")

    print()
    appears = API_KEY in text
    print(f"  the secret appears anywhere in that output: {appears}")
    print(f"  the placeholder appears: {RedactingFilter.PLACEHOLDER in text}")

    print()
    print("Four of the five routes are closed. Read the fifth one carefully,")
    print("because it is the honest limit of this technique:")
    exc_lines = [json.loads(line) for line in text.splitlines()
                 if json.loads(line)["level"] == "ERROR"]
    still_there = API_KEY in json.dumps(exc_lines[0])
    print(f"  the secret survives inside the traceback field: {still_there}")
    print()
    print("The filter rewrites `record.msg`, `record.args` and the fields you")
    print("passed through `extra=`. It does NOT rewrite the traceback, because")
    print("the traceback is rendered later, by the FORMATTER, out of the")
    print("exc_info tuple — after every filter has already run. A secret that")
    print("is inside an exception message therefore walks straight past a")
    print("filter that only touches the record.")
    print()
    print("There are two fixes and one rule.")
    print("  Fix one: scrub in the formatter as well as the filter, so the")
    print("           rendered traceback is scrubbed too.")
    print("  Fix two: never put a credential in an exception message. This is")
    print("           the better fix, because it removes the secret from the")
    print("           exception object rather than from one of its renderings.")
    print("  The rule: a redacting filter is a seatbelt. It is not permission")
    print("           to drive at a wall. Do not log the secret.")


def demo_where_the_filter_goes() -> None:
    banner("2b. WHERE the filter goes, and the hole nobody expects")

    print("The filter above is attached to the HANDLER. The obvious")
    print("alternative is to attach it to the application's top logger and")
    print("let the whole tree inherit the protection. That does not work, and")
    print("here is the measurement rather than the claim.")
    print()

    log, stream = build_logger("leak.demo", secrets=[API_KEY], on_logger=True)
    log.info("logged directly on leak.demo: key=%s", API_KEY)
    child = logging.getLogger("leak.demo.loader")
    child.handlers.clear()
    child.filters.clear()
    child.setLevel(logging.NOTSET)
    child.propagate = True
    child.info("logged on the CHILD leak.demo.loader: key=%s", API_KEY)

    for line in sanitize(stream.getvalue()).splitlines():
        print(f"  {line}")
    print()
    lines = stream.getvalue().splitlines()
    print(f"  the direct line leaked: {API_KEY in lines[0]}")
    print(f"  the child's line leaked: {API_KEY in lines[1]}")
    print()
    print("A logger's filters run only for records logged through THAT logger")
    print("object. A record that arrives by propagation from a descendant")
    print("skips every ancestor's filters — propagation consults the")
    print("ancestors' HANDLERS, not their filters. Since every module in a")
    print("well-behaved application calls getLogger(__name__) and is therefore")
    print("a descendant, a filter on the top logger protects almost nothing")
    print("while looking like it protects everything.")
    print()
    print("Attach redaction to each HANDLER. A handler sees everything that")
    print("reaches its destination, from anywhere in the tree.")

    log, stream = build_logger("safe.demo", secrets=[API_KEY], on_logger=False)
    child = logging.getLogger("safe.demo.loader")
    child.handlers.clear()
    child.filters.clear()
    child.setLevel(logging.NOTSET)
    child.propagate = True
    child.info("logged on the CHILD safe.demo.loader: key=%s", API_KEY)
    print()
    print(f"  {sanitize(stream.getvalue()).strip()}")
    print(f"  the child's line leaked: {API_KEY in stream.getvalue()}")


def demo_formatter_scrubbing() -> None:
    banner("3. Closing the traceback hole: scrub in the formatter too")

    class ScrubbingJsonFormatter(JsonFormatter):
        """JsonFormatter that runs the redactor over the finished line.

        This is the belt to the filter's braces. It works because it happens
        last: by the time `format` returns, the traceback has already been
        rendered into text, so scrubbing the text catches it.
        """

        def __init__(self, secrets, **kwargs):
            super().__init__(**kwargs)
            self._redactor = RedactingFilter(secrets)

        def format(self, record: logging.LogRecord) -> str:
            return self._redactor.scrub(super().format(record))

    logger = logging.getLogger("redact.formatter")
    logger.handlers.clear()
    logger.filters.clear()
    logger.setLevel(logging.DEBUG)
    logger.propagate = False
    handler, stream = buffer_handler(
        logging.DEBUG, ScrubbingJsonFormatter([API_KEY], static_fields={"run_id": RUN_ID})
    )
    logger.addHandler(handler)

    try:
        raise RuntimeError(f"upstream rejected key {API_KEY}")
    except RuntimeError:
        logger.exception("request failed")

    text = stream.getvalue()
    print(f"  the secret appears anywhere in that output: {API_KEY in text}")
    print(f"  the traceback is still there and still useful: "
          f"{'RuntimeError' in text}")
    print()
    print("  the first 200 characters of the line:")
    print(f"  {sanitize(text)[:200]}")
    print()
    print("Cost of this belt: every log line now runs a substring search per")
    print("known secret. That is cheap, and it is not free. The reason to do")
    print("it anyway is that the failure it prevents is not proportional to")
    print("its cost.")


def main() -> None:
    demo_json()
    demo_redaction()
    demo_where_the_filter_goes()
    demo_formatter_scrubbing()
    print()
    print("=" * 70)
    print("Structured logging demonstration complete.")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Four layers of configuration, resolved once, with the provenance of every value.

    python3 examples/04_config_resolver.py

The layers, lowest precedence first:

    default  <  config file  <  environment  <  command-line flag

Five demonstrations:

    1. one setting given a DIFFERENT value in all four layers at once, so you
       can watch each layer override the last and see which one wins
    2. the provenance table: every setting, its value, and where it came from
    3. the type problem — everything from the environment is text, and
       bool("false") is True
    4. missing and empty are different, and the difference is visible
    5. the startup validator, refusing bad values by name and by provenance

Everything is the standard library: os, tomllib, argparse, pathlib.
"""

from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from appconfig import (  # noqa: E402
    APP_SPEC,
    ConfigError,
    Setting,
    resolve,
    to_bool,
    validate,
    validate_or_die,
)

CONFIG_FILE = Path(__file__).resolve().parent / "config.toml"


def banner(title: str) -> None:
    print()
    print("=" * 70)
    print(title)
    print("=" * 70)


def demo_precedence() -> None:
    banner("1. One setting, four different values, four layers")

    print("batch_size is:")
    print("    32   in the code, as the default")
    print("    64   in examples/config.toml")
    print("   128   in the environment, as APP_BATCH_SIZE")
    print("   256   on the command line, as --batch-size")
    print()
    print("Adding one layer at a time:")
    print()

    steps = [
        ("nothing but the code", [], {}, None),
        ("+ the config file", [], {}, CONFIG_FILE),
        ("+ the environment", [], {"APP_BATCH_SIZE": "128"}, CONFIG_FILE),
        ("+ the flag", ["--batch-size", "256"], {"APP_BATCH_SIZE": "128"}, CONFIG_FILE),
    ]
    for label, argv, environ, path in steps:
        config = resolve(APP_SPEC, argv=argv, environ=environ, config_path=path)
        resolved = config.settings["batch_size"]
        print(f"  {label:<24}  batch_size = {resolved.value:<5} from {resolved.source}")

    print()
    print("The flag wins. That ordering is not arbitrary and it is worth being")
    print("able to justify: each layer is more specific than the one below it.")
    print("A default is what everybody gets. A file is what this deployment")
    print("gets. An environment variable is what this process gets. A flag is")
    print("what THIS INVOCATION gets, typed by a person who is looking at the")
    print("problem right now. The more specific statement wins, which is the")
    print("same rule CSS uses and the same rule your shell uses.")


def demo_provenance() -> None:
    banner("2. The provenance table: why is it doing that?")

    config = resolve(
        APP_SPEC,
        argv=["--batch-size", "256", "--log-level", "DEBUG"],
        environ={"APP_SEED": "7", "APP_API_KEY": "sk-live-9f2c4a7b1e63"},
        config_path=CONFIG_FILE,
    )
    print(config.provenance_table())
    print()
    print("Every setting reports its value AND the layer that supplied it.")
    print("Seven settings, five different provenances, one screen. The value of")
    print("this is entirely in the third column: without it, 'why is batch_size")
    print("256?' means reading a TOML file, a deployment manifest, a shell")
    print("wrapper and an argparse definition, in the dark, at speed.")
    print()
    print("Note api_key. It is in the table so you can see WHERE it came from")
    print("and whether it is set at all, and its value is never printed. Those")
    print("are two different questions and only one of them is dangerous.")
    print()
    print("Note also that api_key has no flag. A secret passed on the command")
    print("line is visible in `ps` to every other user on the machine and lands")
    print("in the shell history file. Environment or a secret manager, and")
    print("nowhere else.")


def demo_types() -> None:
    banner("3. Everything from the environment is a string")

    print("The trap, in one line of Python:")
    print(f'    bool("false")  ->  {bool("false")}')
    print()
    print("Every non-empty string is truthy. So the naive conversion turns the")
    print("word 'false' into on, silently, and the feature you switched off")
    print("stays switched on. No error, no warning, just the wrong behaviour.")
    print()
    print("The fix is an explicit table of words, and a refusal for anything")
    print("else:")
    for text in ["true", "TRUE", "1", "yes", "on", "false", "0", "no", "off"]:
        print(f"    to_bool({text!r:<8}) -> {to_bool(text)}")
    for text in ["maybe", "", "2"]:
        try:
            to_bool(text)
        except ValueError as error:
            print(f"    to_bool({text!r:<8}) -> refused: {error}")

    print()
    print("Then the same discipline for the other types. An int setting read")
    print("from the environment:")
    for text in ["128", " 128 ", "12.5"]:
        try:
            config = resolve(APP_SPEC, argv=[], environ={"APP_BATCH_SIZE": text})
            print(f"    APP_BATCH_SIZE={text!r:<8} -> {config['batch_size']}")
        except ConfigError as error:
            print(f"    APP_BATCH_SIZE={text!r:<8} -> refused: {error}")

    print()
    print("And the difference the config file makes: TOML has real types, so")
    print("`batch_size = 64` in the file arrives as an int already and needs no")
    print("conversion at all. That is a genuine advantage of a typed file")
    print("format over the environment, and it is the reason a file is a better")
    print("home for structured configuration than a pile of variables.")
    config = resolve(APP_SPEC, argv=[], environ={}, config_path=CONFIG_FILE)
    print(f"    from config.toml: batch_size = {config['batch_size']!r} "
          f"({type(config['batch_size']).__name__}), "
          f"dry_run = {config['dry_run']!r} "
          f"({type(config['dry_run']).__name__})")


def demo_missing_versus_empty() -> None:
    banner("4. A missing variable and an empty one are different")

    print("These are three different states of one environment variable, and a")
    print("program that cannot tell them apart will eventually do the wrong")
    print("thing with at least one:")
    print()

    cases = [
        ("not set at all", {}),
        ("set to the empty string", {"APP_MODEL_NAME": ""}),
        ("set to a value", {"APP_MODEL_NAME": "large-encoder"}),
    ]
    for label, environ in cases:
        config = resolve(APP_SPEC, argv=[], environ=environ, config_path=CONFIG_FILE)
        resolved = config.settings["model_name"]
        print(f"  {label:<26} value={resolved.value!r:<18} source={resolved.source}")

    print()
    print("The distinction is made by asking `'APP_MODEL_NAME' in environ`")
    print("rather than `os.environ.get('APP_MODEL_NAME')`. `.get` returns None")
    print("for a variable that was never set and '' for one that was set to")
    print("nothing, and the usual `or default` idiom then collapses both to the")
    print("default:")
    print()
    print('    name = os.environ.get("APP_MODEL_NAME") or "tiny-baseline"')
    print()
    print("That line cannot express 'the operator deliberately blanked this'.")
    print("It matters because an empty variable is almost never an accident —")
    print("it is a deployment template that filled in nothing, a secret that")
    print("failed to inject, or a person who meant to clear a value. Silently")
    print("treating it as 'unset' hides all three.")
    print()
    print("For an int setting the empty string is not a value at all, and the")
    print("resolver says so rather than guessing:")
    try:
        resolve(APP_SPEC, argv=[], environ={"APP_BATCH_SIZE": ""})
    except ConfigError as error:
        print(f"    {error}")


def demo_validation() -> None:
    banner("5. Validating at startup, so nothing fails at 3 a.m.")

    print("A bad configuration value has two possible moments of discovery:")
    print("the second the process starts, or the first time the code path that")
    print("uses it runs — which may be hours later, in the middle of the night,")
    print("halfway through a job. Validation at startup chooses the first.")
    print()

    config = resolve(
        APP_SPEC,
        argv=["--batch-size", "0", "--log-level", "VERBOSE"],
        environ={"APP_SEED": "-1"},
        config_path=CONFIG_FILE,
    )
    problems = validate(config, APP_SPEC)
    print(f"  {len(problems)} problems found, all of them at once:")
    for problem in problems:
        print(f"    - {problem}")

    print()
    print("Two design decisions in those messages.")
    print()
    print("They are reported ALL AT ONCE rather than one per run, because")
    print("fixing configuration one error at a time is miserable and pushes")
    print("people towards guessing.")
    print()
    print("And every one names its PROVENANCE. 'batch_size must be at least 1'")
    print("tells you what is wrong. 'batch_size: 0 is below the minimum of 1")
    print("(from flag:--batch-size)' tells you where to go and change it. The")
    print("second message costs one extra field on a dataclass.")
    print()
    print("A good configuration then passes silently:")
    good = resolve(
        APP_SPEC,
        argv=["--batch-size", "128"],
        environ={"APP_SEED": "7"},
        config_path=CONFIG_FILE,
    )
    validate_or_die(good, APP_SPEC)
    print("    validate_or_die() returned; the program may start.")

    print()
    print("And a required secret that is absent is a configuration error too,")
    print("not a runtime surprise. Adding one rule to the spec:")
    required_key = Setting(
        name="api_key", kind="str", default="", env="APP_API_KEY",
        flag=None, secret=True, help="credential",
    )
    empty = resolve([required_key], argv=[], environ={})
    if not empty["api_key"]:
        print("    api_key: not set. Set APP_API_KEY in the environment.")
        print("    (the message names the variable and never the value)")


def main() -> None:
    demo_precedence()
    demo_provenance()
    demo_types()
    demo_missing_versus_empty()
    demo_validation()
    print()
    print("=" * 70)
    print("Configuration demonstration complete.")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""dictConfig, two handlers with two formatters, and file rotation.

    python3 examples/05_dictconfig_and_rotation.py [WORKDIR]

Three things:

    1. `logging.config.dictConfig` — the configuration form worth knowing,
       because it puts the whole logging setup in ONE dictionary that can
       itself come from a config file, which is the point of the day
    2. `RotatingFileHandler` and `TimedRotatingFileHandler`, demonstrated
       until they actually rotate
    3. the honest note: for a service, stdout plus a supervisor is usually
       the better answer, and Day 81 and Day 84 both argued this already

If WORKDIR is not given the script makes a temporary directory and removes it
on the way out, so it leaves nothing behind.
"""

from __future__ import annotations

import json
import logging
import logging.config
import logging.handlers
import shutil
import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

RUN_ID = "run-4711"
API_KEY = "sk-live-9f2c4a7b1e63"  # invented for this lab; not a real credential


def banner(title: str) -> None:
    print()
    print("=" * 70)
    print(title)
    print("=" * 70)


def build_dict_config(log_dir: Path, level: str) -> dict:
    """The whole logging setup as one dictionary.

    Read it top to bottom: it is the four objects of the logging module,
    written out as four sections. Formatters render. Filters decide and edit.
    Handlers send somewhere, each with its own level and its own formatter.
    Loggers are named and say which handlers they use.

    `disable_existing_loggers` is the setting nobody reads and everybody is
    bitten by. It defaults to True, which silences every logger that already
    existed when this call is made — including the ones created at import
    time by libraries you depend on. Set it to False unless you specifically
    want that.
    """
    return {
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            # For a person, on a terminal.
            "console": {
                "format": "%(asctime)s %(levelname)-8s %(name)-14s %(message)s",
                "datefmt": "%H:%M:%S",
            },
            # For a machine, in a file. Same records, different rendering:
            # the formatter belongs to the handler, not to the record.
            "json": {
                "()": "applog.JsonFormatter",
                "static_fields": {"run_id": RUN_ID},
            },
        },
        "filters": {
            "redact": {
                "()": "applog.RedactingFilter",
                "secrets": [API_KEY],
            },
        },
        "handlers": {
            "console": {
                "class": "logging.StreamHandler",
                "level": "INFO",           # people want the summary
                "formatter": "console",
                "filters": ["redact"],
                "stream": "ext://sys.stdout",
            },
            "file": {
                "class": "logging.handlers.RotatingFileHandler",
                "level": "DEBUG",          # the file keeps everything
                "formatter": "json",
                "filters": ["redact"],
                "filename": str(log_dir / "app.log"),
                "maxBytes": 900,           # absurdly small, to force rotation
                "backupCount": 3,
                "encoding": "utf-8",
            },
        },
        "loggers": {
            "app": {
                "level": level,            # the LOGGER's level: the first gate
                "handlers": ["console", "file"],
                "propagate": False,
            },
        },
        "root": {"level": "WARNING", "handlers": ["console"]},
    }


def demo_dictconfig(log_dir: Path) -> logging.Logger:
    banner("1. dictConfig: the whole setup in one dictionary")

    config = build_dict_config(log_dir, level="DEBUG")
    logging.config.dictConfig(config)
    log = logging.getLogger("app.prepare")

    print("Configured. One logger, two handlers, two formatters, one filter.")
    print("Watch the SAME five calls produce two different renderings.")
    print()
    print("--- what the console handler prints (level INFO, human format) ---")
    log.debug("cache hit for shard 3")
    log.info("run started")
    log.info("batch complete", extra={"batch": 1, "kept": 61})
    log.warning("upstream slow", extra={"status": 429})
    log.info("using key %s", API_KEY)
    print()

    lines = (log_dir / "app.log").read_text(encoding="utf-8").splitlines()
    print("--- what the file handler wrote (level DEBUG, JSON) ---")
    for line in lines:
        print(f"  {line}")
    print()
    print(f"console saw 4 of the 5 calls; the file has {len(lines)} of them.")
    print("The DEBUG line was accepted by the logger, rejected by the console")
    print("handler and kept by the file handler — the two-level rule, with two")
    print("handlers disagreeing on purpose, which is what it is FOR.")
    print()
    key_in_file = API_KEY in (log_dir / "app.log").read_text(encoding="utf-8")
    print(f"the secret appears in the file: {key_in_file}")
    print("The redacting filter is listed on BOTH handlers, which is the only")
    print("arrangement that actually works: a filter on the `app` logger would")
    print("be skipped by every record propagating up from `app.prepare`, and")
    print("both destinations would be leaking. Demonstration 03 measures that.")

    print()
    print("Why a dictionary rather than calls: this is DATA. It can be loaded")
    print("from the TOML file the rest of the configuration comes from, kept in")
    print("version control, diffed in review, and swapped per environment")
    print("without touching a line of Python. `logging.basicConfig` is the")
    print("convenience version of the same thing — one handler on the ROOT")
    print("logger, and nothing at all if the root already has one, which is")
    print("why calling it twice appears to do nothing the second time.")
    return log


def demo_rotation(log_dir: Path) -> None:
    banner("2. Rotation: what happens when the file gets big")

    log = logging.getLogger("app.rotate")
    # DEBUG, so these 40 lines go to the file and not to your terminal. The
    # console handler is at INFO; the file handler is at DEBUG.
    for index in range(40):
        log.debug("processing record", extra={"record": index})

    handler = logging.getLogger("app").handlers[1]
    handler.flush()

    files = sorted(p.name for p in log_dir.glob("app.log*"))
    print(f"maxBytes=900, backupCount=3. After 40 more records:")
    for name in files:
        path = log_dir / name
        print(f"  {name:<12} {path.stat().st_size:>6} bytes, "
              f"{len(path.read_text(encoding='utf-8').splitlines())} lines")
    print()
    print("`app.log` is always the live one. When it exceeds maxBytes it is")
    print("renamed to `app.log.1`, the old `.1` becomes `.2`, and the file")
    print("that would have become `.4` is DELETED — backupCount is how many")
    print("you keep, and everything past it is gone for good.")
    print()
    print("`TimedRotatingFileHandler` is the same idea keyed on the clock")
    print("rather than the size. Waiting for midnight is impractical in a")
    print("demonstration, so here it is with the rollover called by hand —")
    print("which is exactly what the handler does when the interval elapses:")

    timed = logging.handlers.TimedRotatingFileHandler(
        filename=str(log_dir / "daily.log"), when="midnight", backupCount=7,
        encoding="utf-8",
    )
    timed.setFormatter(logging.Formatter("%(message)s"))
    timed_log = logging.getLogger("timed.demo")
    timed_log.handlers.clear()
    timed_log.propagate = False
    timed_log.setLevel(logging.INFO)
    timed_log.addHandler(timed)

    timed_log.info("yesterday's work")
    timed.doRollover()
    timed_log.info("today's work")
    timed.close()

    daily = sorted(p.name for p in log_dir.glob("daily.log*"))
    print(f"  files after one rollover: {len(daily)}")
    for name in daily:
        suffix = "the live file" if name == "daily.log" else "rolled, named by date"
        print(f"    {name:<24} {suffix}")
    print()
    print("Size-based rotation bounds your DISK; time-based rotation bounds")
    print("your SEARCH, because 'the log for Tuesday' is one file. Pick by")
    print("which of those two questions you ask more often. `backupCount=7`")
    print("with when='midnight' is a week of history and no more.")

    print()
    print("Now the honest part, and it is the same conclusion Day 81 reached")
    print("about scheduling and Day 84 reached about packaging.")
    print()
    print("For a long-running service, writing your own log files is usually")
    print("the WRONG default. Log to stdout, unformatted by any rotation")
    print("logic, and let the thing that already supervises your process")
    print("collect it — systemd's journal, a container runtime, a process")
    print("manager. Four reasons, none of them theoretical:")
    print()
    print("  * Rotation in-process is racy across processes. Two workers with")
    print("    RotatingFileHandler on the same file will rename it out from")
    print("    under each other. There is a documented recipe involving a")
    print("    socket handler and a single writer, and it exists because the")
    print("    simple thing does not work.")
    print("  * The supervisor already does it, uniformly, for every service on")
    print("    the machine, with one retention policy you can audit.")
    print("  * A container has no persistent disk to rotate onto anyway.")
    print("  * stdout composes. `your-app | grep ERROR` works; a file handler")
    print("    inside the process does not compose with anything.")
    print()
    print("When file rotation IS right: a scheduled job that runs on a machine")
    print("with a disk and no supervisor, a desktop application, or a")
    print("deliberately separate audit trail with its own retention rules.")
    print("Those are real cases. They are not the default.")


def demo_level_change(log_dir: Path) -> None:
    banner("3. Turning it down without editing code")

    print("The whole argument for logging over print, in one demonstration.")
    print("Same program, same calls, one configuration value changed:")
    print()
    for level in ("DEBUG", "INFO", "WARNING"):
        for path in log_dir.glob("app.log*"):
            path.unlink()
        logging.config.dictConfig(build_dict_config(log_dir, level=level))
        log = logging.getLogger("app.levels")
        log.debug("cache hit for shard 3")
        log.info("batch complete", extra={"batch": 1})
        log.warning("upstream slow", extra={"status": 429})
        logging.getLogger("app").handlers[1].flush()
        lines = (log_dir / "app.log").read_text(encoding="utf-8").splitlines()
        events = [json.loads(line)["event"] for line in lines]
        print(f"  logger level {level:<8} -> file has {len(lines)} lines: {events}")
    print()
    print("Three deployments, three answers, one unchanged program. With")
    print("print() the only way to get the first row is to edit the source and")
    print("ship it, and the only way to get back to the third is to edit it")
    print("again — which is exactly the change nobody wants to make while")
    print("something is broken.")


def main() -> None:
    if len(sys.argv) > 1:
        log_dir, temporary = Path(sys.argv[1]), False
        log_dir.mkdir(parents=True, exist_ok=True)
    else:
        log_dir, temporary = Path(tempfile.mkdtemp(prefix="day097-")), True

    try:
        demo_dictconfig(log_dir)
        demo_rotation(log_dir)
        demo_level_change(log_dir)
        print()
        print("=" * 70)
        print("dictConfig and rotation demonstration complete.")
        if temporary:
            print("(the log directory was temporary and has been removed)")
        else:
            print(f"(log files left in {log_dir.name}/ because you named a directory)")
    finally:
        logging.shutdown()
        if temporary:
            shutil.rmtree(log_dir, ignore_errors=True)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""The two halves joined: a run that can be reconstructed from its own record.

    python3 examples/06_run_manifest.py
    python3 examples/06_run_manifest.py --seed 11 --batch-size 128 --run-id run-4712

A training run you cannot reconstruct is an anecdote, not a result. This
script is the smallest honest version of the thing that makes a run
reproducible:

    * the CONFIGURATION is resolved through the four layers, and every value
      knows which layer it came from
    * the MANIFEST — model, data version, seed, batch size, and the
      provenance of each — is written into the log as the first event, so
      the log answers "what was this run actually configured with?" without
      anybody having to remember
    * the RUN LOG is JSON with a run_id on every line, so two runs of the
      same program are separable
    * the API KEY is present in the configuration, used by the program, and
      appears in neither the manifest nor any log line

Nothing here trains anything. The arithmetic is a deterministic stand-in, so
that the same seed produces the same numbers and you can see for yourself
that the log is enough to reproduce the run.
"""

from __future__ import annotations

import json
import logging
import random
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from appconfig import APP_SPEC, ConfigError, resolve, validate_or_die  # noqa: E402
from applog import JsonFormatter, RedactingFilter  # noqa: E402

CONFIG_FILE = Path(__file__).resolve().parent / "config.toml"


def configure_logging(run_id: str, level: str, secrets: list[str]) -> logging.Logger:
    """One JSON handler on stdout, with the run id stamped on every line.

    stdout, not a file. Whatever supervises this process — a scheduler, a
    container runtime, a CI job — already collects stdout and already has a
    retention policy. Writing a file here would mean owning rotation, disk
    space and cleanup for no benefit. Day 81 and Day 84 both landed on this.
    """
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JsonFormatter(static_fields={"run_id": run_id}))
    # On the HANDLER. A filter on a logger is skipped by every record that
    # propagates up from a child logger, which is every module in the program.
    handler.addFilter(RedactingFilter(secrets))

    logger = logging.getLogger("run")
    logger.handlers.clear()
    logger.filters.clear()
    logger.setLevel(getattr(logging, level))
    logger.addHandler(handler)
    logger.propagate = False
    return logger


def fake_step(rng: random.Random, step: int) -> float:
    """A deterministic stand-in for a training step. Same seed, same numbers."""
    return round(2.0 / (step + 1) + rng.random() * 0.01, 6)


def main(argv: list[str] | None = None) -> int:
    argv = sys.argv[1:] if argv is None else argv

    # The run id is configuration too, and it defaults to something fixed here
    # so the captured output is reproducible. In a real job it would be the
    # scheduler's job id, the CI run number, or a uuid4 — anything that is
    # unique and that you can paste into a search box.
    run_id = "run-4711"
    if "--run-id" in argv:
        index = argv.index("--run-id")
        run_id = argv[index + 1]
        argv = argv[:index] + argv[index + 2:]

    try:
        config = resolve(APP_SPEC, argv=argv, config_path=CONFIG_FILE)
        validate_or_die(config, APP_SPEC)
    except ConfigError as error:
        # Configuration failures are reported before logging is configured,
        # on stderr, and the process stops. There is nothing useful to do
        # with a program that does not know what it is meant to do.
        print(f"configuration error:\n{error}", file=sys.stderr)
        return 2

    secrets = [config["api_key"]] if config["api_key"] else []
    log = configure_logging(run_id, config["log_level"], secrets)

    # THE MANIFEST. First line of the run, and the reason the run is
    # reconstructable. Note safe_dict(): api_key is in the configuration and
    # is not in this event.
    log.info(
        "run started",
        extra={
            "config": config.safe_dict(),
            "provenance": {name: r.source for name, r in config.settings.items()},
        },
    )

    rng = random.Random(config["seed"])
    losses = []
    for step in range(1, 4):
        loss = fake_step(rng, step)
        losses.append(loss)
        log.info("step complete", extra={"step": step, "loss": loss})

    if config["dry_run"]:
        log.warning("dry run: nothing was written", extra={"artifact": None})
    else:
        log.info("artifact written", extra={"artifact": f"{config['model_name']}.bin"})

    log.info("run finished", extra={"steps": len(losses), "final_loss": losses[-1]})

    # The plain-language summary goes to stderr so that stdout stays a clean
    # stream of JSON objects. Two streams, two audiences, no interleaving.
    print("", file=sys.stderr)
    print("--- the same run, as a person would read it (stderr) ---", file=sys.stderr)
    print(f"run {run_id}: model={config['model_name']} "
          f"data={config['data_version']} seed={config['seed']} "
          f"batch={config['batch_size']}", file=sys.stderr)
    print(f"  final loss {losses[-1]}", file=sys.stderr)
    print(f"  api_key configured: {bool(config['api_key'])} "
          f"(value never printed, never logged)", file=sys.stderr)
    print("", file=sys.stderr)
    print("To repeat this run exactly, read the manifest out of its own log:",
          file=sys.stderr)
    print("  python3 examples/06_run_manifest.py \\", file=sys.stderr)
    print(f"      --seed {config['seed']} --batch-size {config['batch_size']} \\",
          file=sys.stderr)
    print(f"      --model-name {config['model_name']} "
          f"--data-version {config['data_version']}", file=sys.stderr)
    print("", file=sys.stderr)
    print("That command is derivable from the log because the log recorded the",
          file=sys.stderr)
    print("configuration. If it had not, the honest answer to 'what produced",
          file=sys.stderr)
    print("this number?' would be 'nobody knows', and the number would be an",
          file=sys.stderr)
    print("anecdote rather than a result.", file=sys.stderr)
    return 0


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

#!/usr/bin/env python3
"""Reference answers to exercises 1-6. Read AFTER you have tried them.

This file has the same public names as `starter/01_logging.py`, so
`starter/03_check.sh` can be pointed at either one. The test suite uses that
to prove the checker is not vacuous: it runs the checker against this file and
requires 6 of 6, then against the untouched starter and requires 0 of 6.

The two classes are imported from `examples/applog.py` rather than written
twice, because they are the same code and a second copy would drift.
"""

from __future__ import annotations

import io
import logging
import sys
from pathlib import Path
from typing import Any, Iterable  # noqa: F401  (kept to mirror the starter)

sys.path.insert(0, str(Path(__file__).resolve().parent))

from applog import JsonFormatter, RedactingFilter, iso_utc  # noqa: E402,F401

SECRET = "sk-live-9f2c4a7b1e63"  # invented for this lab; not a real credential

VALID_LABELS = {"neutral", "negative", "positive"}

RECORDS = [
    {"id": 1, "text": "the cat sat on the mat", "label": "neutral"},
    {"id": 2, "text": "", "label": "neutral"},
    {"id": 3, "text": "shipping was late and the box was crushed", "label": "negative"},
    {"id": 4, "text": "arrived early, works perfectly", "label": "positive"},
    {"id": 5, "text": "no opinion", "label": "unknown"},
    {"id": 6, "text": "great value for the price", "label": "positive"},
]


def buffer_handler(level: int, formatter: logging.Formatter):
    stream = io.StringIO()
    handler = logging.StreamHandler(stream)
    handler.setLevel(level)
    handler.setFormatter(formatter)
    return handler, stream


# --- EXERCISE 1 ------------------------------------------------------------
# getLogger, not Logger(). getLogger returns the SAME object for the same
# name every time, from a module-level registry, which is what makes
# configuring "myapp" configure "myapp.loader" too. Constructing a Logger
# directly bypasses that registry and the object is connected to nothing.
log = logging.getLogger(__name__)


# --- EXERCISE 2 ------------------------------------------------------------
def prepare(records: list[dict], logger: logging.Logger) -> list[dict]:
    logger.info("preparation starting: %d records", len(records))
    kept = []
    for record in records:
        logger.debug("processing record %s", record["id"])
        if not record["text"]:
            logger.warning("skipping record %s: empty text", record["id"])
            continue
        if record["label"] not in VALID_LABELS:
            logger.warning(
                "skipping record %s: unknown label %r", record["id"], record["label"]
            )
            continue
        kept.append(record)
    logger.info("preparation done: kept %d of %d", len(kept), len(records))
    return kept


# --- EXERCISE 3 ------------------------------------------------------------
def make_debug_logger(name: str, handler: logging.Handler) -> logging.Logger:
    logger = logging.getLogger(name)
    logger.handlers.clear()
    logger.filters.clear()
    logger.setLevel(logging.DEBUG)     # gate one: the logger
    handler.setLevel(logging.DEBUG)    # gate two: the handler
    logger.addHandler(handler)
    logger.propagate = False
    return logger


# --- EXERCISE 4 ------------------------------------------------------------
def parse_batch_size(text: str) -> int:
    return int(text)


def log_parse_failure(text: str, logger: logging.Logger) -> bool:
    try:
        parse_batch_size(text)
    except ValueError:
        # exception() is error() with exc_info=True. It reads the exception
        # currently being handled out of the interpreter, so it takes no
        # argument — and it must be called inside the except block.
        logger.exception("could not parse batch size")
        return False
    return True


# --- EXERCISES 5 and 6 -----------------------------------------------------
# Both are in examples/applog.py, imported above and re-exported here under
# the names the checker looks for.
STANDARD_KEYS = frozenset(
    vars(
        logging.LogRecord(
            name="", level=0, pathname="", lineno=0, msg="", args=(), exc_info=None
        )
    )
) | {"message", "asctime", "taskName"}


def main() -> None:
    import json

    handler, stream = buffer_handler(logging.DEBUG, JsonFormatter({"run_id": "run-1"}))
    handler.addFilter(RedactingFilter([SECRET]))
    logger = make_debug_logger("scratch", handler)
    prepare(RECORDS, logger)
    log_parse_failure("sixty-four", logger)
    for line in stream.getvalue().splitlines():
        record = json.loads(line)
        record.pop("traceback", None)
        print(json.dumps(record))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Reference answers to exercises 7-12. Read AFTER you have tried them.

Same public names as `starter/02_config.py`, so `starter/03_check.sh` can be
pointed at either one. The resolver itself lives in `examples/appconfig.py`;
this file supplies the names the checker looks for and the SPEC the starter
uses.
"""

from __future__ import annotations

import argparse  # noqa: F401  (kept to mirror the starter)
import os
import sys
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parent))

from appconfig import (  # noqa: E402,F401
    FALSE_WORDS,
    TRUE_WORDS,
    Config,
    ConfigError,
    Resolved,
    Setting,
    build_parser,
    load_toml,
    resolve,
    to_bool,
    validate,
)

# The same five settings the starter file specifies.
SPEC: tuple[Setting, ...] = (
    Setting("log_level", "str", "INFO", env="APP_LOG_LEVEL", flag="--log-level",
            choices=("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")),
    Setting("batch_size", "int", 32, env="APP_BATCH_SIZE", flag="--batch-size",
            minimum=1, maximum=1024),
    Setting("model_name", "str", "tiny-baseline", env="APP_MODEL_NAME",
            flag="--model-name"),
    Setting("dry_run", "bool", False, env="APP_DRY_RUN", flag="--dry-run"),
    Setting("api_key", "str", "", env="APP_API_KEY", flag=None, secret=True),
)


def safe_dict(config: Config) -> dict[str, Any]:
    """The configuration with every secret replaced. The only loggable version."""
    return config.safe_dict()


def main() -> None:
    config = resolve(SPEC, argv=sys.argv[1:], environ=dict(os.environ))
    print(config.provenance_table())
    for problem in validate(config, SPEC):
        print(f"PROBLEM: {problem}")


if __name__ == "__main__":
    main()

"""A four-layer configuration resolver that remembers where every value came from.

The four layers, lowest precedence first:

    1. defaults       written in the code, so the program runs with none of
                      the other three present
    2. config file    TOML, read with `tomllib` from the standard library
    3. environment    os.environ, which is where secrets and per-deployment
                      values belong
    4. command line   argparse flags, which are the most specific and
                      therefore win

Every setting carries a `Resolved` record — the value AND the layer that
supplied it AND the raw text before conversion. That is the whole reason
this module exists. "Why is it doing that?" should be answered by printing
the configuration, in five seconds, not by reading four files and guessing.

Everything here is the standard library: `os`, `tomllib`, `argparse`,
`pathlib`, `dataclasses`.
"""

from __future__ import annotations

import argparse
import os
import tomllib
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Sequence

# --------------------------------------------------------------------------
# Type conversion. Everything from a file, an environment variable or a flag
# arrives as text, and text is not a type.
# --------------------------------------------------------------------------

TRUE_WORDS = frozenset({"1", "true", "yes", "on"})
FALSE_WORDS = frozenset({"0", "false", "no", "off"})


class ConfigError(Exception):
    """Raised at startup when configuration is unusable. Never at 3 a.m."""


def to_bool(text: str) -> bool:
    """Convert text to a bool, refusing anything ambiguous.

    This function exists because `bool("false")` is `True`. Every non-empty
    string is truthy in Python, so the naive version of this conversion
    turns "false" into on, silently, and the feature you switched off stays
    switched on. There is no warning; there is no error; there is only the
    wrong behaviour.
    """
    lowered = text.strip().lower()
    if lowered in TRUE_WORDS:
        return True
    if lowered in FALSE_WORDS:
        return False
    raise ValueError(
        f"expected one of {sorted(TRUE_WORDS | FALSE_WORDS)}, got {text!r}"
    )


def to_int(text: str) -> int:
    return int(text.strip())


def to_str(text: str) -> str:
    return text


CONVERTERS: dict[str, Callable[[str], Any]] = {
    "str": to_str,
    "int": to_int,
    "bool": to_bool,
}


# --------------------------------------------------------------------------
# The specification of one setting, and the record of one resolved value.
# --------------------------------------------------------------------------


@dataclass(frozen=True)
class Setting:
    """What a setting is called in each of the four layers, and its rules."""

    name: str
    kind: str                       # "str" | "int" | "bool"
    default: Any
    env: str | None = None          # environment variable name
    flag: str | None = None         # command-line flag, e.g. "--batch-size"
    choices: tuple[Any, ...] | None = None
    minimum: int | None = None
    maximum: int | None = None
    secret: bool = False            # never printed, never logged
    help: str = ""


@dataclass(frozen=True)
class Resolved:
    """One setting's value, and the provenance of that value."""

    name: str
    value: Any
    source: str                     # "default" | "file:..." | "env:..." | "flag:..."
    raw: str | None = None          # the text before conversion, when there was any
    secret: bool = False

    def display(self) -> str:
        return "***redacted***" if self.secret and self.value is not None else repr(self.value)


@dataclass
class Config:
    """The resolved configuration: values by name, each with its provenance."""

    settings: dict[str, Resolved] = field(default_factory=dict)

    def __getitem__(self, name: str) -> Any:
        return self.settings[name].value

    def source_of(self, name: str) -> str:
        return self.settings[name].source

    def as_dict(self) -> dict[str, Any]:
        return {name: r.value for name, r in self.settings.items()}

    def safe_dict(self) -> dict[str, Any]:
        """Every value except the secrets, which become the placeholder.

        This is the dictionary you are allowed to log.
        """
        return {
            name: ("***redacted***" if r.secret and r.value is not None else r.value)
            for name, r in self.settings.items()
        }

    def provenance_table(self) -> str:
        """The five-second answer to "why is it doing that?"."""
        width_n = max(len(n) for n in self.settings)
        width_v = max(len(r.display()) for r in self.settings.values())
        width_n = max(width_n, len("setting"))
        width_v = max(width_v, len("value"))
        lines = [
            f"{'setting'.ljust(width_n)}  {'value'.ljust(width_v)}  came from",
            f"{'-' * width_n}  {'-' * width_v}  {'-' * 24}",
        ]
        for name in self.settings:
            r = self.settings[name]
            lines.append(f"{name.ljust(width_n)}  {r.display().ljust(width_v)}  {r.source}")
        return "\n".join(lines)


# --------------------------------------------------------------------------
# The resolver.
# --------------------------------------------------------------------------


def load_toml(path: Path | None) -> tuple[dict[str, Any], str | None]:
    """Read a TOML file if it exists. Returns the table and the path used.

    `tomllib` has been in the standard library since Python 3.11, and it is
    READ-ONLY on purpose — there is no `tomllib.dump`. If your program needs
    to write TOML you need a third-party library; if it only needs to read
    its own configuration, which is the overwhelmingly common case, the
    standard library is enough.

    Note the mode: `tomllib.load` requires a binary file object. Passing a
    text file raises `TypeError`, and that catches everybody once.
    """
    if path is None or not path.exists():
        return {}, None
    with path.open("rb") as handle:
        return tomllib.load(handle), str(path.name)


def build_parser(spec: Sequence[Setting]) -> argparse.ArgumentParser:
    """One flag per setting that has one. Defaults are deliberately absent.

    argparse's own `default=` is not used, because then a value that came
    from a default would be indistinguishable from one the user typed. The
    parser's job here is only to report what was actually passed; the
    layering is done by this module.
    """
    parser = argparse.ArgumentParser(
        prog="app",
        description="Demonstration application with four-layer configuration.",
        add_help=True,
    )
    parser.add_argument("--config", default=None, help="path to a TOML config file")
    for setting in spec:
        if setting.flag is None:
            continue
        if setting.kind == "bool":
            # Two flags rather than one, so --no-x can override a file or an
            # environment variable that said true. A lone --x can only ever
            # turn something on, which makes the highest-precedence layer
            # unable to express half of the values.
            parser.add_argument(
                setting.flag, dest=setting.name, action="store_const",
                const="true", default=None, help=setting.help,
            )
            parser.add_argument(
                setting.flag.replace("--", "--no-", 1), dest=setting.name,
                action="store_const", const="false", default=None,
                help=f"disable {setting.name}",
            )
        else:
            parser.add_argument(
                setting.flag, dest=setting.name, default=None, help=setting.help
            )
    return parser


def resolve(
    spec: Sequence[Setting],
    argv: Sequence[str] | None = None,
    environ: dict[str, str] | None = None,
    config_path: Path | None = None,
) -> Config:
    """Resolve every setting through the four layers, recording provenance.

    `argv` and `environ` are parameters with defaults rather than reads of
    `sys.argv` and `os.environ`, for exactly the reason Day 91's report took
    its instant as a parameter: a function that reaches out to global state
    cannot be tested, and configuration resolution is the one piece of code
    you most want to be able to test.
    """
    environ = os.environ if environ is None else environ
    parser = build_parser(spec)
    args = parser.parse_args([] if argv is None else list(argv))

    if args.config is not None:
        config_path = Path(args.config)
    file_table, file_name = load_toml(config_path)

    config = Config()
    for setting in spec:
        # ---- layer 1: the default in the code -----------------------------
        value, source, raw = setting.default, "default", None

        # ---- layer 2: the config file -------------------------------------
        if setting.name in file_table:
            file_value = file_table[setting.name]
            # TOML has real types, so a value read from it is already an int
            # or a bool. Only convert when it arrived as text.
            if isinstance(file_value, str) and setting.kind != "str":
                raw = file_value
                value = _convert(setting, file_value, f"file:{file_name}")
            else:
                raw = None
                value = _typecheck(setting, file_value, f"file:{file_name}")
            source = f"file:{file_name}"

        # ---- layer 3: the environment -------------------------------------
        if setting.env is not None and setting.env in environ:
            text = environ[setting.env]
            # A missing variable and an empty one are DIFFERENT. `in environ`
            # asks whether it was set at all; the empty string is a value
            # somebody chose. `os.environ.get(name)` collapses the two into
            # None-or-text and loses the distinction, which is why this uses
            # containment rather than .get().
            if text == "":
                value = "" if setting.kind == "str" else setting.default
                source = f"env:{setting.env} (set but empty)"
                raw = ""
                if setting.kind != "str":
                    raise ConfigError(
                        f"{setting.name}: environment variable {setting.env} is set "
                        f"but empty, and an empty string is not a valid "
                        f"{setting.kind}. Unset it to use the default, or give it a value."
                    )
            else:
                raw = text
                value = _convert(setting, text, f"env:{setting.env}")
                source = f"env:{setting.env}"

        # ---- layer 4: the command line ------------------------------------
        typed = getattr(args, setting.name, None)
        if typed is not None:
            raw = typed
            value = _convert(setting, typed, f"flag:{setting.flag}")
            source = f"flag:{setting.flag}"

        config.settings[setting.name] = Resolved(
            name=setting.name, value=value, source=source, raw=raw, secret=setting.secret
        )
    return config


def _convert(setting: Setting, text: str, where: str) -> Any:
    try:
        return CONVERTERS[setting.kind](text)
    except ValueError as error:
        raise ConfigError(
            f"{setting.name}: cannot read {text!r} from {where} as {setting.kind} ({error})"
        ) from error


def _typecheck(setting: Setting, value: Any, where: str) -> Any:
    expected = {"str": str, "int": int, "bool": bool}[setting.kind]
    # bool is a subclass of int in Python, so an int setting must reject True
    # explicitly or `batch_size = true` in the TOML file becomes 1.
    if setting.kind == "int" and isinstance(value, bool):
        raise ConfigError(f"{setting.name}: {where} gave a boolean where an int was expected")
    if not isinstance(value, expected):
        raise ConfigError(
            f"{setting.name}: {where} gave {type(value).__name__} "
            f"where {setting.kind} was expected"
        )
    return value


# --------------------------------------------------------------------------
# Validation, run once at startup.
# --------------------------------------------------------------------------


def validate(config: Config, spec: Sequence[Setting]) -> list[str]:
    """Return every problem, each naming the setting AND where the value came from.

    Two deliberate choices.

    It returns ALL the problems rather than raising on the first one, because
    fixing configuration one error per run is miserable and encourages people
    to guess.

    Every message names the provenance. "batch_size must be at least 1" tells
    you what is wrong. "batch_size: 0 is below the minimum of 1 (from
    flag:--batch-size)" tells you where to go and fix it.
    """
    problems: list[str] = []
    for setting in spec:
        resolved = config.settings[setting.name]
        value, where = resolved.value, resolved.source
        shown = "***redacted***" if setting.secret else repr(value)

        if setting.choices is not None and value not in setting.choices:
            problems.append(
                f"{setting.name}: {shown} is not one of "
                f"{list(setting.choices)} (from {where})"
            )
        if setting.minimum is not None and isinstance(value, int) and value < setting.minimum:
            problems.append(
                f"{setting.name}: {shown} is below the minimum of "
                f"{setting.minimum} (from {where})"
            )
        if setting.maximum is not None and isinstance(value, int) and value > setting.maximum:
            problems.append(
                f"{setting.name}: {shown} is above the maximum of "
                f"{setting.maximum} (from {where})"
            )
    return problems


def validate_or_die(config: Config, spec: Sequence[Setting]) -> None:
    """Fail at startup, loudly, with every problem listed at once."""
    problems = validate(config, spec)
    if problems:
        raise ConfigError(
            "configuration is not usable:\n  - " + "\n  - ".join(problems)
        )


# --------------------------------------------------------------------------
# The specification this lab's demonstration application uses.
# --------------------------------------------------------------------------

APP_SPEC: tuple[Setting, ...] = (
    Setting(
        name="log_level", kind="str", default="INFO",
        env="APP_LOG_LEVEL", flag="--log-level",
        choices=("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"),
        help="lowest severity that reaches the handlers",
    ),
    Setting(
        name="batch_size", kind="int", default=32,
        env="APP_BATCH_SIZE", flag="--batch-size",
        minimum=1, maximum=1024,
        help="records processed per batch",
    ),
    Setting(
        name="model_name", kind="str", default="tiny-baseline",
        env="APP_MODEL_NAME", flag="--model-name",
        help="which model this run uses",
    ),
    Setting(
        name="seed", kind="int", default=0,
        env="APP_SEED", flag="--seed", minimum=0,
        help="random seed, so the run can be repeated",
    ),
    Setting(
        name="dry_run", kind="bool", default=False,
        env="APP_DRY_RUN", flag="--dry-run",
        help="do everything except write anything",
    ),
    Setting(
        name="data_version", kind="str", default="v1",
        env="APP_DATA_VERSION", flag="--data-version",
        help="which snapshot of the dataset this run read",
    ),
    # No flag. A secret passed on the command line is visible in the process
    # table to every other user on the machine, and lands in the shell
    # history file. Environment or a secret manager, and nowhere else.
    Setting(
        name="api_key", kind="str", default="",
        env="APP_API_KEY", flag=None, secret=True,
        help="credential for the upstream service (environment only)",
    ),
)

"""Reusable logging pieces: a JSON formatter and a redacting filter.

This is the "from scratch" half of the day's logging material. Both classes
are small on purpose — the point is that the four objects the `logging`
module gives you (logger, handler, formatter, filter) are separate, and that
you extend the two of them that are meant to be extended.

Nothing here is clever. `JsonFormatter` overrides one method, `format`.
`RedactingFilter` overrides one method, `filter`. Everything else is
inherited.
"""

from __future__ import annotations

import json
import logging
import time
from typing import Any, Iterable

# Attributes the logging module puts on every LogRecord. Anything on a record
# that is NOT in this set was put there by the caller through `extra=`, which
# is exactly the material a structured log wants to carry.
#
# Built by asking the logging module rather than by hand-copying its
# documentation, so it cannot drift out of date with the interpreter you are
# actually running.
_STANDARD_RECORD_KEYS = frozenset(
    vars(
        logging.LogRecord(
            name="", level=0, pathname="", lineno=0, msg="", args=(), exc_info=None
        )
    )
) | {"message", "asctime", "taskName"}


def iso_utc(epoch_seconds: float) -> str:
    """Format an epoch timestamp as ISO 8601 in UTC, to milliseconds.

    Day 91 made the argument for this format in a database: fixed-width,
    most-significant-first, so text order is chronological order. A log file
    wants the same property, for the same reason — `sort` on a log file
    should put the lines in the order the events happened.
    """
    whole = int(epoch_seconds)
    millis = int(round((epoch_seconds - whole) * 1000))
    if millis == 1000:  # a rounding edge; roll it into the next second
        whole += 1
        millis = 0
    return time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(whole)) + f".{millis:03d}Z"


class JsonFormatter(logging.Formatter):
    """Render a LogRecord as one JSON object on one line.

    Why one line: every log-shipping tool ever written reads a stream of
    lines. A multi-line JSON document is a parsing problem for the reader;
    a one-line JSON object is not.

    The fixed fields are the ones a person searching at 3 a.m. filters on:
    when, how bad, who said it, what happened. Everything passed through
    `extra=` is merged in beside them, so `log.info("step done",
    extra={"step": 3})` produces a `step` field you can query on rather than
    a number buried in prose.
    """

    def __init__(self, static_fields: dict[str, Any] | None = None) -> None:
        super().__init__()
        # Fields stamped onto every record — the run id belongs here, because
        # a value that must appear on every line should be attached once
        # rather than remembered at every call site.
        self.static_fields = dict(static_fields or {})

    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, Any] = {
            "ts": iso_utc(record.created),
            "level": record.levelname,
            "logger": record.name,
            "event": record.getMessage(),
        }
        payload.update(self.static_fields)

        for key, value in vars(record).items():
            if key in _STANDARD_RECORD_KEYS or key.startswith("_"):
                continue
            payload[key] = value

        if record.exc_info:
            # formatException gives the same text `logging.exception` would
            # have printed. It goes in a field of its own rather than being
            # glued onto the message, so the parser does not have to guess
            # where the message ends and the traceback starts.
            payload["exc_type"] = record.exc_info[0].__name__
            payload["traceback"] = self.formatException(record.exc_info)

        # default=str so an unexpected object in `extra=` degrades to its
        # repr instead of raising inside the logging call. A logging call
        # that raises is a bug that only shows up when something is already
        # going wrong, which is the worst possible time.
        return json.dumps(payload, default=str, sort_keys=False)


class RedactingFilter(logging.Filter):
    """Replace known secret VALUES anywhere in a record with a placeholder.

    **Attach it to every HANDLER, not to a logger.** This is the single most
    important line in this file and it was learned the hard way while writing
    this lab. A filter attached to a logger runs only for records logged
    through *that logger object*. Records that arrive by propagation from a
    descendant logger — `app.loader` when the filter is on `app` — skip the
    ancestor's filters entirely; propagation consults the ancestors' HANDLERS,
    not their filters. So `logging.getLogger("app").addFilter(redactor)` looks
    like whole-application protection and is not: every `getLogger(__name__)`
    in every module is a descendant, and every one of them bypasses it.
    Demonstration 03 shows this happening. A filter on each handler sees
    everything that reaches that destination, which is what you actually want.

    The design decision worth arguing about: this redacts by value, not by
    key name. Key-name redaction ("hide anything called password") misses
    `log.info("calling %s", url_with_token)`, which is how secrets actually
    escape. Value redaction catches the secret wherever it appears — in the
    message, in the arguments, in an `extra` field.

    The honest limits, and they are real:

    * It only knows the values you hand it. A secret it has never been told
      about goes straight through.
    * It cannot see a secret that has been transformed — base64-encoded,
      truncated, or split across two log calls.
    * Values shorter than `min_length` are ignored, because redacting every
      occurrence of a two-character secret would destroy the log.

    A redacting filter is a seatbelt, not a reason to drive at a wall. The
    rule stays: do not log the secret.
    """

    PLACEHOLDER = "***redacted***"

    def __init__(self, secrets: Iterable[str], min_length: int = 6) -> None:
        super().__init__()
        self.min_length = min_length
        self.secrets = sorted(
            {s for s in secrets if s and len(s) >= min_length},
            key=len,
            reverse=True,  # longest first, so a secret containing another
        )                  # secret is not half-redacted into readability

    def scrub(self, text: str) -> str:
        for secret in self.secrets:
            if secret in text:
                text = text.replace(secret, self.PLACEHOLDER)
        return text

    def _scrub_value(self, value: Any) -> Any:
        if isinstance(value, str):
            return self.scrub(value)
        if isinstance(value, (list, tuple)):
            return type(value)(self._scrub_value(v) for v in value)
        if isinstance(value, dict):
            return {k: self._scrub_value(v) for k, v in value.items()}
        return value

    def filter(self, record: logging.LogRecord) -> bool:
        # A filter returns True to keep the record and False to drop it.
        # Mutating the record on the way through is explicitly allowed by the
        # logging documentation, and is what makes redaction possible at all.
        if isinstance(record.msg, str):
            record.msg = self.scrub(record.msg)
        if record.args:
            record.args = self._scrub_value(record.args)
        for key, value in list(vars(record).items()):
            if key in _STANDARD_RECORD_KEYS or key.startswith("_"):
                continue
            setattr(record, key, self._scrub_value(value))
        return True


def buffer_handler(
    level: int = logging.DEBUG, formatter: logging.Formatter | None = None
) -> tuple[logging.Handler, Any]:
    """A handler that writes into a StringIO, plus the StringIO itself.

    This is how the tests capture log output, and how you should capture it
    too. Scraping stdout means asserting on whatever else the program
    printed; a buffer handler receives exactly the records that reached this
    handler and nothing else.
    """
    import io

    stream = io.StringIO()
    handler = logging.StreamHandler(stream)
    handler.setLevel(level)
    handler.setFormatter(formatter or JsonFormatter())
    return handler, stream

# Layer 2 of four: the config file.
#
# This file holds the values a whole team shares and that are safe to commit:
# which model, which data snapshot, how big a batch. It holds NO secret. A
# secret in a file in a repository is a secret in everybody's laptop, in every
# clone, in every backup, and in the history forever after somebody deletes it.
#
# TOML is read by `tomllib` from the standard library (Python 3.11+). Note that
# TOML has real types — 64 below is an integer, not the string "64" — so a
# value read from here needs no conversion, which is exactly the difference
# between this layer and the environment.

batch_size = 64
model_name = "small-encoder"
data_version = "2026-08-01"
dry_run = false

# log_level is deliberately absent, so the resolver has to fall back to the
# default for it. A setting that appears in no file, no environment variable
# and no flag must still have a value, or the program cannot start.

# seed is deliberately absent too, for the same reason and because a run that
# does not state its seed is a run nobody can repeat.

lesson_id: D097
day: 97
kind: guided-build
languages: [python, bash]
setup_commands:
  - cd labs/sections/programming-with-python/day-097-logging-and-configuration
  - python3 --version
  - python3 -c "import tomllib, logging.config; print('standard library ready')"
run_commands:
  - bash tests/run_tests.sh
  - bash starter/03_check.sh
  - python3 examples/01_prints.py
  - python3 examples/02_logging_architecture.py
  - python3 examples/03_structured_logging.py
  - python3 examples/04_config_resolver.py
  - python3 examples/05_dictconfig_and_rotation.py
  - APP_API_KEY=sk-live-9f2c4a7b1e63 APP_SEED=7 python3 examples/06_run_manifest.py
  - bash starter/03_check.sh examples/07_solution_logging.py examples/08_solution_config.py
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - find . -type d -name __pycache__ -prune -exec rm -rf -- {} +
  - 'git checkout -- starter/  # optional: reset your work'
requires_network: false
requires_api_key: false
estimated_minutes: 30
last_executed: '2026-08-16'
executed_on: 'macOS 26.5.2 (Apple Silicon, arm64), Python 3.14.0, bash 3.2.57 — bash tests/run_tests.sh -> 86 checks, 0 failure(s), exit 0'

# Dependencies

**None.** This lab installs nothing, and `requirements.txt` is deliberately
empty of packages. That is not minimalism for its own sake: the argument of
the day is that the standard library already contains a complete logging
framework and everything you need to resolve configuration, and you cannot
judge `structlog` or `pydantic-settings` fairly until you have built the small
version yourself.

| Tool | Version used here | Where it comes from | Licence |
| --- | --- | --- | --- |
| `python3` | 3.14.0 | Whatever Python you installed on Day 43 | PSF licence |
| `bash` | 3.2.57 | Preinstalled on macOS and every Linux distribution | GPL |

Modules used, all from the standard library: `logging`, `logging.config`,
`logging.handlers`, `json`, `os`, `sys`, `tomllib`, `argparse`, `pathlib`,
`dataclasses`, `random`, `tempfile`, `shutil`, `time`, `io`, `importlib.util`,
`subprocess`, `re`.

Check what you have:

```bash
python3 --version
python3 -c "import tomllib, logging.config, logging.handlers; print('ready')"
```

## Minimum version, and why

**Python 3.11 or newer**, for `tomllib`. That is the only floor this lab has,
and it is a real one: `tomllib` was added in 3.11 (2022), and the configuration
half reads its file layer with it.

If you are on 3.10 or older, `pip install tomli` gives you the same parser
under the name `tomli` — `tomllib` was adopted from it, so the code is the
same. Everything else here has been in the standard library far longer:
`logging` and `logging.config` since Python 2.3 (2003), `dictConfig` since
Python 2.7 and 3.2 (2010), `argparse` since 3.2.

The `str | Path` type hints and the `dict[str, Any]` builtin generics also
want 3.10 or newer, and `from __future__ import annotations` at the top of
each file keeps them from being evaluated at import time.

## If your Python is somewhere unusual

Both scripts take an override rather than guessing:

```bash
PYTHON=/path/to/python3 bash tests/run_tests.sh
PYTHON=/path/to/python3 bash starter/03_check.sh
```

They fail loudly with that instruction if they cannot find one, rather than
quietly skipping the checks that need it.

## What is deliberately absent

**No `structlog`, no `loguru`, no `python-json-logger`.** All three are good,
all three are free, and all three are covered in the lesson's Alternatives
section from their documentation. None of them is installed on the machine
this lab was captured on, so **no output is reproduced for any of them** — the
lesson says so plainly where it discusses each one. The JSON formatter you
build in exercise 5 is about forty lines and is what `python-json-logger`
does; writing it once is how you find out that a formatter is a class with one
method.

**No `pydantic-settings`, no `dynaconf`.** Same treatment: described, not
demonstrated, because neither is installed here. Day 94 covered pydantic
itself, and `pydantic-settings` is the natural next step for a real service —
after you have written the resolver by hand and know what it is doing for you.

**`python-dotenv` is the one exception**, and it is worth being precise about
why. It is not used by this lab and is not needed by it. It does, however,
happen to be present in the system interpreter this lab was captured on, at
version 1.2.2 — pulled in as a dependency of something else installed on that
machine. Because it was genuinely available, the lesson runs it once and shows
the real output, clearly marked as an aside. You do not need it to complete
anything here, and nothing in the lab imports it.

**No log-shipping platform.** The lesson names the category and stops there.
Which collector or hosted service you use is an operations decision with
prices attached, it changes yearly, and none of it changes the four objects
inside your process. What travels is the JSON line, and you have written the
thing that produces it.

# Day 097 — Logging and Configuration
#
# This lab has no third-party dependencies. It uses python3 with the standard
# library only: logging, logging.config, logging.handlers, json, os, tomllib,
# argparse, pathlib, dataclasses, random and tempfile.
#
# There is nothing to install. See requirements/README.md for the versions used
# when the expected output was captured, for the single version floor this lab
# does have (tomllib, Python 3.11), and for which third-party logging and
# configuration libraries are deliberately absent — and which one happened to
# be present and is therefore shown running.

# The brief — "Say It Where Someone Will Read It"

Read this before you write anything.

## The situation

You have a script. It works. It is full of `print`.

It runs on your laptop while you watch it, and every line it prints is useful
to you, right now, at your desk. `examples/01_prints.py` is that script — run
it once and read what comes out.

Tomorrow it moves to a machine you do not have a terminal on. It runs at
04:00. It runs again at 05:00. Its output lands in a file that also holds
last week's output. Somebody who is not you will read that file, in a hurry,
because something has gone wrong.

Every one of those `print` lines is now useless, and each for a different
reason:

| What the line says | Why it stops working |
| --- | --- |
| `processing record 3` | Which run? There are 168 of them in that file this week |
| `skipping record 2: empty text` | When? There is no timestamp on any line |
| `could not write output` | How bad? Same shape as the line above it, so no alert can tell them apart |
| all of them | How do you turn them down? You edit the source and deploy |
| `using API key sk-live-...` | This one is a security incident, not an inconvenience |

That is the whole day, in one sentence: **`print` is for you, at your desk,
right now; logging is for whoever is awake when it breaks.**

## What you are building

Twelve exercises across two files.

`starter/01_logging.py` — the logging half.

1. A module-level logger, obtained the way every module should obtain one.
2. The `print`-based function converted, with each line given the level it
   deserves.
3. A handler whose level does not silently swallow what the logger accepted —
   the two-level trap, met and fixed.
4. A failure logged with `exception()` rather than `error(str(e))`.
5. A JSON formatter, so the log becomes a table you can query.
6. A redacting filter, so a known secret cannot reach any handler.

`starter/02_config.py` — the configuration half.

7. `to_bool`, which refuses to believe that the string `"false"` is true.
8. The four-layer resolver: default, then file, then environment, then flag.
9. Provenance: every setting reports where its value came from.
10. Missing and empty environment variables, told apart.
11. A startup validator that names the setting and the layer it came from.
12. `safe_dict`, which is the configuration with the secrets removed — the
    only version that may be logged.

## How to work

```bash
# from the lab directory
bash starter/03_check.sh
```

It will say `0 of 12 exercises complete.` and, for each one, what it wanted
and what it got. Work down the two files, re-running it as you go. It exits
non-zero until all twelve pass.

The checker never reads how you wrote something. It runs your code and looks
at the values, so any correct implementation passes.

## The rules of the exercise

- **The standard library only.** `logging`, `logging.config`, `os`,
  `tomllib`, `argparse`, `json`, `pathlib`. Nothing to install.
- **Capture logs with a handler writing to a buffer**, never by scraping
  stdout. `examples/applog.py` has `buffer_handler` if you want to see one.
- **The secret is `sk-live-9f2c4a7b1e63`**, invented for this lab. Exercise 6
  is not complete until that string appears nowhere in the captured output.
- **Read `examples/` after you have tried, not before.** Every exercise has a
  worked reference there, and reading it first costs you the exercise.

## The one thing to carry out of the day

Both halves are the same idea wearing different clothes.

Logging is how a program tells you what it did. Configuration is how you tell
a program what to do. Neither should require editing the source, and both
should be able to answer "why?" without anybody guessing — which is why the
resolver you build in exercise 9 records provenance and the log you build in
exercise 5 records a run id.

#!/usr/bin/env python3
"""EXERCISES 1-6 — the logging half. Your work goes here.

Check your progress at any time:

    bash starter/03_check.sh

Everything below either works already or raises NotImplementedError with the
exercise number in the message. Nothing is a stub: the two pieces that are
written for you are complete and worth reading, because they show the shape
the rest should take.

Standard library only. `logging` and `json` are already imported.
"""

from __future__ import annotations

import io
import json
import logging
import time
from typing import Any, Iterable

SECRET = "sk-live-9f2c4a7b1e63"  # invented for this lab; not a real credential

VALID_LABELS = {"neutral", "negative", "positive"}

RECORDS = [
    {"id": 1, "text": "the cat sat on the mat", "label": "neutral"},
    {"id": 2, "text": "", "label": "neutral"},
    {"id": 3, "text": "shipping was late and the box was crushed", "label": "negative"},
    {"id": 4, "text": "arrived early, works perfectly", "label": "positive"},
    {"id": 5, "text": "no opinion", "label": "unknown"},
    {"id": 6, "text": "great value for the price", "label": "positive"},
]


# ---------------------------------------------------------------------------
# WRITTEN FOR YOU — read this before starting. It is how the checker captures
# your log output, and it is how you should capture log output in a test.
# ---------------------------------------------------------------------------

def buffer_handler(level: int, formatter: logging.Formatter) -> tuple[logging.Handler, io.StringIO]:
    """A handler that writes into a StringIO, plus the StringIO itself."""
    stream = io.StringIO()
    handler = logging.StreamHandler(stream)
    handler.setLevel(level)
    handler.setFormatter(formatter)
    return handler, stream


def iso_utc(epoch_seconds: float) -> str:
    """ISO 8601 in UTC to milliseconds. Written for you; exercise 5 uses it."""
    whole = int(epoch_seconds)
    millis = int(round((epoch_seconds - whole) * 1000))
    if millis == 1000:
        whole += 1
        millis = 0
    return time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(whole)) + f".{millis:03d}Z"


# ---------------------------------------------------------------------------
# EXERCISE 1 — a module-level logger
#
# Replace the None below with the one-line idiom every module in every Python
# program should use to obtain its logger. It must be:
#
#   * obtained from the logging module rather than constructed
#   * named after THIS module, so that the dotted hierarchy works and
#     configuring the parent name configures everything beneath it
#
# The checker asserts that `log.name` equals this module's __name__ and that
# `log` is the same object `logging.getLogger` returns for that name.
# ---------------------------------------------------------------------------

log = None  # EXERCISE 1: one line, and the answer is in the paragraph above


# ---------------------------------------------------------------------------
# EXERCISE 2 — convert the print-based function
#
# `examples/01_prints.py` has the original. Rewrite it here using `logger`
# instead of print, with these severities and no others:
#
#   * "preparation starting: N records"          -> INFO
#   * "processing record N"                      -> DEBUG   (per-record noise)
#   * "skipping record N: empty text"            -> WARNING (survivable)
#   * "skipping record N: unknown label 'x'"     -> WARNING
#   * "preparation done: kept M of N"            -> INFO
#
# Two requirements the checker enforces:
#
#   * use LAZY formatting — logger.info("kept %d of %d", m, n) — and NOT an
#     f-string. The checker inspects `record.msg` and fails if the number has
#     already been baked into the template.
#   * RETURN the kept records. The result of the function is not log output.
#
# The API key must not appear anywhere in this function. It never was the
# log's business.
# ---------------------------------------------------------------------------

def prepare(records: list[dict], logger: logging.Logger) -> list[dict]:
    raise NotImplementedError(
        "EXERCISE 2: convert examples/01_prints.py's prepare() to logging calls"
    )


# ---------------------------------------------------------------------------
# EXERCISE 3 — the two-level trap
#
# Return a logger set up so that a DEBUG call actually comes out.
#
# The trap: a record must pass the LOGGER's level AND then the HANDLER's
# level, and they are two different objects. Setting the logger to DEBUG and
# leaving the handler at WARNING drops every debug line with no error, and
# is the most common logging question there is.
#
# Build a logger named `name` that:
#   * accepts DEBUG at the logger
#   * has exactly one handler, which also accepts DEBUG
#   * does NOT propagate, so the checker's buffer is the only destination
#   * uses the handler and formatter you are given
#
# Return the logger. The caller keeps the stream.
# ---------------------------------------------------------------------------

def make_debug_logger(name: str, handler: logging.Handler) -> logging.Logger:
    raise NotImplementedError(
        "EXERCISE 3: both the logger's level AND the handler's level must pass"
    )


# ---------------------------------------------------------------------------
# EXERCISE 4 — log a failure so somebody can fix it
#
# Call `parse_batch_size(text)` inside a try block. When it raises ValueError,
# log it at ERROR with the message "could not parse batch size" AND the
# traceback attached.
#
# `logger.error(str(error))` loses the traceback, which is the whole value:
# it is the only thing that says which line, in which function, called with
# what. There is a one-word method that does the right thing inside an
# except block, and it takes no exception argument.
#
# Return True if the parse succeeded, False if it was logged as a failure.
# The checker asserts record.exc_info is not None and that the formatted
# output contains the word "Traceback".
# ---------------------------------------------------------------------------

def parse_batch_size(text: str) -> int:
    """Written for you. It raises ValueError on anything that is not an int."""
    return int(text)


def log_parse_failure(text: str, logger: logging.Logger) -> bool:
    raise NotImplementedError(
        "EXERCISE 4: use the method that attaches the traceback automatically"
    )


# ---------------------------------------------------------------------------
# EXERCISE 5 — a JSON formatter
#
# Fill in `format` so each record becomes ONE line of JSON with these keys:
#
#   ts      iso_utc(record.created)   — the helper above does the formatting
#   level   the level NAME, not the number
#   logger  the logger's name
#   event   the FORMATTED message (there is a record method for this; using
#           record.msg would give you the template with %s still in it)
#
# then every key from `self.static_fields`, then every attribute the caller
# added through `extra=`. `STANDARD_KEYS` below tells you which attributes
# the logging module put there itself, so anything NOT in it came from the
# caller and belongs in the output.
#
# The checker parses your lines with json.loads and asserts on the values,
# so key order does not matter and whitespace does not matter.
# ---------------------------------------------------------------------------

STANDARD_KEYS = frozenset(
    vars(
        logging.LogRecord(
            name="", level=0, pathname="", lineno=0, msg="", args=(), exc_info=None
        )
    )
) | {"message", "asctime", "taskName"}


class JsonFormatter(logging.Formatter):
    def __init__(self, static_fields: dict[str, Any] | None = None) -> None:
        super().__init__()
        self.static_fields = dict(static_fields or {})

    def format(self, record: logging.LogRecord) -> str:
        raise NotImplementedError(
            "EXERCISE 5: return one line of JSON with ts, level, logger, event, "
            "the static fields, and everything passed through extra="
        )


# ---------------------------------------------------------------------------
# EXERCISE 6 — a redacting filter
#
# Fill in `filter` so that no known secret VALUE survives on the record.
#
# A filter returns True to keep the record and False to drop it. Mutating the
# record on the way through is allowed, and is what makes redaction possible.
# Three places a secret hides, and the checker tests all three:
#
#   * record.msg      — the message itself, when somebody used an f-string
#   * record.args     — the arguments of a lazy-formatted call
#   * the attributes the caller added through extra=, INCLUDING values nested
#     inside a dict, because {"headers": {"Authorization": "Bearer sk-..."}}
#     is exactly how this happens in real code
#
# Replace each occurrence with self.PLACEHOLDER. Return True.
#
# Where this filter gets ATTACHED matters more than how it is written, and
# the answer is not the obvious one — attach it to each HANDLER. See
# examples/03_structured_logging.py demonstration 2b for the measurement.
# ---------------------------------------------------------------------------

class RedactingFilter(logging.Filter):
    PLACEHOLDER = "***redacted***"

    def __init__(self, secrets: Iterable[str]) -> None:
        super().__init__()
        self.secrets = [s for s in secrets if s]

    def filter(self, record: logging.LogRecord) -> bool:
        raise NotImplementedError(
            "EXERCISE 6: scrub record.msg, record.args and every extra= field, "
            "including values nested inside a dict"
        )


# ---------------------------------------------------------------------------
# A place to try things out. `python3 starter/01_logging.py` runs this.
# ---------------------------------------------------------------------------

def main() -> None:
    handler, stream = buffer_handler(logging.DEBUG, JsonFormatter({"run_id": "run-1"}))
    handler.addFilter(RedactingFilter([SECRET]))
    logger = make_debug_logger("scratch", handler)
    prepare(RECORDS, logger)
    log_parse_failure("sixty-four", logger)
    for line in stream.getvalue().splitlines():
        print(json.dumps(json.loads(line), indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""EXERCISES 7-12 — the configuration half. Your work goes here.

Check your progress at any time:

    bash starter/03_check.sh

The `Setting`, `Resolved` and `Config` types are written for you and are
complete. What is missing is the resolution itself, and it is the part worth
writing by hand once.

Standard library only: os, tomllib, argparse, pathlib, dataclasses.
"""

from __future__ import annotations

import argparse
import os
import tomllib
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Sequence

TRUE_WORDS = frozenset({"1", "true", "yes", "on"})
FALSE_WORDS = frozenset({"0", "false", "no", "off"})


class ConfigError(Exception):
    """Raised at startup when configuration is unusable."""


# ---------------------------------------------------------------------------
# WRITTEN FOR YOU — the three types. Read them; they define the shape of the
# answer. The important one is Resolved: a value AND its provenance, together,
# because they are useless apart.
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class Setting:
    name: str
    kind: str                       # "str" | "int" | "bool"
    default: Any
    env: str | None = None
    flag: str | None = None
    choices: tuple[Any, ...] | None = None
    minimum: int | None = None
    maximum: int | None = None
    secret: bool = False


@dataclass(frozen=True)
class Resolved:
    name: str
    value: Any
    source: str                     # "default" | "file:..." | "env:..." | "flag:..."
    raw: str | None = None
    secret: bool = False


@dataclass
class Config:
    settings: dict[str, Resolved] = field(default_factory=dict)

    def __getitem__(self, name: str) -> Any:
        return self.settings[name].value

    def source_of(self, name: str) -> str:
        return self.settings[name].source


# The specification the checker uses. Do not change it.
SPEC: tuple[Setting, ...] = (
    Setting("log_level", "str", "INFO", env="APP_LOG_LEVEL", flag="--log-level",
            choices=("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")),
    Setting("batch_size", "int", 32, env="APP_BATCH_SIZE", flag="--batch-size",
            minimum=1, maximum=1024),
    Setting("model_name", "str", "tiny-baseline", env="APP_MODEL_NAME",
            flag="--model-name"),
    Setting("dry_run", "bool", False, env="APP_DRY_RUN", flag="--dry-run"),
    # No flag, deliberately: a secret on the command line is visible in `ps`
    # to every other user on the machine and lands in the shell history file.
    Setting("api_key", "str", "", env="APP_API_KEY", flag=None, secret=True),
)


# ---------------------------------------------------------------------------
# WRITTEN FOR YOU — the argument parser. Note what it does NOT do: it sets no
# argparse defaults, so a flag that was not typed comes back as None and is
# distinguishable from one that was. The layering is your job, not argparse's.
# ---------------------------------------------------------------------------

def build_parser(spec: Sequence[Setting]) -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="app", add_help=True)
    for setting in spec:
        if setting.flag is None:
            continue
        if setting.kind == "bool":
            parser.add_argument(setting.flag, dest=setting.name,
                                action="store_const", const="true", default=None)
            parser.add_argument(setting.flag.replace("--", "--no-", 1),
                                dest=setting.name, action="store_const",
                                const="false", default=None)
        else:
            parser.add_argument(setting.flag, dest=setting.name, default=None)
    return parser


# ---------------------------------------------------------------------------
# EXERCISE 7 — to_bool
#
# `bool("false")` is True. Every non-empty string is truthy in Python, so the
# naive conversion turns the word "false" into on, silently, and the feature
# you switched off stays switched on.
#
# Return True for anything in TRUE_WORDS, False for anything in FALSE_WORDS,
# case-insensitively and ignoring surrounding whitespace. Raise ValueError for
# anything else — including "", "maybe" and "2". Guessing is what got us here.
# ---------------------------------------------------------------------------

def to_bool(text: str) -> bool:
    raise NotImplementedError("EXERCISE 7: an explicit table, and a refusal for the rest")


# ---------------------------------------------------------------------------
# EXERCISE 8 — read the TOML file
#
# Return (table, filename) for a file that exists, and ({}, None) for a path
# that is None or missing. `tomllib.load` needs a BINARY file object; passing
# a text one raises TypeError, and that catches everybody exactly once.
#
# tomllib has been in the standard library since Python 3.11 and is read-only:
# there is no tomllib.dump. That is a deliberate scope decision, and it is
# fine, because a program almost always reads its configuration and almost
# never writes it.
# ---------------------------------------------------------------------------

def load_toml(path: Path | None) -> tuple[dict[str, Any], str | None]:
    raise NotImplementedError("EXERCISE 8: open it in binary mode, and handle 'missing'")


# ---------------------------------------------------------------------------
# EXERCISES 9, 10, 11 — the resolver
#
# Fill in `resolve` so that every setting is resolved through four layers,
# lowest precedence first:
#
#   1. default          source "default"
#   2. config file      source "file:<filename>"
#   3. environment      source "env:<VARIABLE>"
#   4. command line     source "flag:<--flag>"
#
# Each layer that has a value overwrites the one before it, and records its
# own name as the provenance. That is EXERCISE 9 (the layering) and EXERCISE
# 11 (the provenance) together, and they are one loop.
#
# EXERCISE 10 is the awkward one, and it is awkward in real life too. A
# missing environment variable and an empty one are DIFFERENT:
#
#   * not set at all              -> fall through to the layer below
#   * set to ""                   -> the operator said something. For a str
#                                    setting the value is "" and the source is
#                                    "env:NAME (set but empty)". For an int or
#                                    a bool setting, raise ConfigError naming
#                                    the variable, because "" is not a number.
#
# Ask `name in environ`, not `environ.get(name)`. `.get` returns None for
# never-set and "" for set-to-empty, and the usual `or default` idiom then
# collapses both into the default and loses the distinction forever.
#
# Types: a value from the file is already an int or a bool, because TOML has
# real types. A value from the environment or a flag is always text and must
# be converted. Conversion failures raise ConfigError naming the setting AND
# where the bad value came from.
# ---------------------------------------------------------------------------

def resolve(
    spec: Sequence[Setting],
    argv: Sequence[str] | None = None,
    environ: dict[str, str] | None = None,
    config_path: Path | None = None,
) -> Config:
    raise NotImplementedError(
        "EXERCISES 9-11: four layers, in order, each recording its own provenance"
    )


# ---------------------------------------------------------------------------
# EXERCISE 12 — validate at startup, and safe_dict
#
# `validate` returns a list of problem strings — ALL of them, not just the
# first, because fixing configuration one error per run is miserable.
#
# Check `choices`, `minimum` and `maximum` where the Setting defines them.
# Every message must contain:
#
#   * the setting's NAME
#   * what is wrong
#   * the PROVENANCE, so the reader knows which of the four places to go and
#     edit. The checker requires the source string to appear in the message.
#
# Never put a secret's VALUE in a message. Use "***redacted***" instead.
#
# `safe_dict` returns {name: value} with every secret replaced by
# "***redacted***". It is the only version of the configuration that may be
# logged, and exercise 5's JSON formatter is where it ends up.
# ---------------------------------------------------------------------------

def validate(config: Config, spec: Sequence[Setting]) -> list[str]:
    raise NotImplementedError(
        "EXERCISE 12: every problem at once, each naming the setting and its source"
    )


def safe_dict(config: Config) -> dict[str, Any]:
    raise NotImplementedError("EXERCISE 12: the configuration, minus the secrets")


# ---------------------------------------------------------------------------
# A place to try things out. `python3 starter/02_config.py --batch-size 256`
# ---------------------------------------------------------------------------

def main() -> None:
    import sys

    config = resolve(SPEC, argv=sys.argv[1:], environ=dict(os.environ))
    problems = validate(config, SPEC)
    for name, resolved in config.settings.items():
        shown = "***redacted***" if resolved.secret and resolved.value else repr(resolved.value)
        print(f"{name:<12} {shown:<18} {resolved.source}")
    for problem in problems:
        print(f"PROBLEM: {problem}")


if __name__ == "__main__":
    main()

#!/usr/bin/env bash
# How far through the twelve exercises are you?
#
#   bash starter/03_check.sh
#
# It imports starter/01_logging.py and starter/02_config.py, calls your
# functions, captures log output into a buffer, and compares real values. It
# never looks at HOW you wrote anything, so any correct implementation passes.
#
# Exit status: 0 when all twelve pass, 1 otherwise.
#
# To check the reference answers instead of your own work:
#   bash starter/03_check.sh examples/07_solution_logging.py examples/08_solution_config.py
set -u

lab_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
python_bin="${PYTHON:-$(command -v python3 || true)}"

if [ -z "${python_bin}" ] || [ ! -x "${python_bin}" ]; then
  echo "python3 was not found. Install Python 3.11+ or set PYTHON=/path/to/python3."
  exit 1
fi

logging_file="${1:-${lab_dir}/starter/01_logging.py}"
config_file="${2:-${lab_dir}/starter/02_config.py}"

export PYTHONDONTWRITEBYTECODE=1

echo "Day 097 — Say It Where Someone Will Read It"
echo "python3: $("${python_bin}" -c 'import sys; print(sys.version.split()[0])')"
echo
echo "Checking:"
echo "  $(basename "$(dirname "${logging_file}")")/$(basename "${logging_file}")"
echo "  $(basename "$(dirname "${config_file}")")/$(basename "${config_file}")"
echo

"${python_bin}" "${lab_dir}/tests/check_exercises.py" "${logging_file}" "${config_file}"
status=$?

if [ "${status}" -ne 0 ]; then
  echo
  echo "Keep going. Each 'not yet' line above says what it wanted and what it got."
  echo "The brief is starter/00_brief.md; the reference answers are in examples/,"
  echo "and are worth more after you have tried than before."
fi
exit "${status}"

#!/usr/bin/env python3
"""Check the twelve exercises by RUNNING them and looking at the values.

Usage (normally through starter/03_check.sh):

    python3 tests/check_exercises.py LOGGING_FILE CONFIG_FILE

It never reads how you wrote something. It imports your two files, calls your
functions, captures log output into a buffer, and compares real values — so
any correct implementation passes and a plausible-looking wrong one does not.

Prints one line per exercise and a final "N of 12 exercises complete.".
Exit status is 0 only when N is 12.
"""

from __future__ import annotations

import importlib.util
import io
import json
import logging
import sys
import tempfile
from pathlib import Path

SECRET = "sk-live-9f2c4a7b1e63"  # invented for this lab; not a real credential

# When a handler or formatter raises, the logging module prints the traceback
# to stderr and carries on — deliberately, because a logging failure must not
# take the program down with it. While the exercises are unfinished that
# happens on every call, so the checker's own output would be buried. Turning
# raiseExceptions off silences it; the checker reports the failure itself.
logging.raiseExceptions = False

results: list[tuple[int, str, bool, str]] = []


def record(number: int, title: str, ok: bool, detail: str = "") -> None:
    results.append((number, title, ok, detail))


def load(path: Path, module_name: str):
    spec = importlib.util.spec_from_file_location(module_name, path)
    if spec is None or spec.loader is None:
        raise ImportError(f"cannot load {path}")
    module = importlib.util.module_from_spec(spec)
    sys.modules[module_name] = module
    spec.loader.exec_module(module)
    return module


def capture(logger_name: str, formatter: logging.Formatter, filters=()):
    stream = io.StringIO()
    handler = logging.StreamHandler(stream)
    handler.setLevel(logging.DEBUG)
    handler.setFormatter(formatter)
    for f in filters:
        handler.addFilter(f)
    logger = logging.getLogger(logger_name)
    logger.handlers.clear()
    logger.filters.clear()
    logger.setLevel(logging.DEBUG)
    logger.propagate = False
    logger.addHandler(handler)
    return logger, stream


class Collector(logging.Handler):
    """Keeps the LogRecord objects themselves, not their rendered text."""

    def __init__(self) -> None:
        super().__init__(level=logging.DEBUG)
        self.records: list[logging.LogRecord] = []

    def emit(self, record: logging.LogRecord) -> None:
        self.records.append(record)


# ---------------------------------------------------------------------------
# Exercises 1-6
# ---------------------------------------------------------------------------


def check_logging(path: Path) -> None:
    try:
        mod = load(path, "starter_logging")
    except Exception as error:  # noqa: BLE001
        for number in range(1, 7):
            record(number, "(module did not import)", False, f"{type(error).__name__}: {error}")
        return

    # 1 --------------------------------------------------------------------
    try:
        ok = (
            isinstance(mod.log, logging.Logger)
            and mod.log.name == mod.__name__
            and mod.log is logging.getLogger(mod.__name__)
        )
        detail = f"log = {mod.log!r}"
    except Exception as error:  # noqa: BLE001
        ok, detail = False, f"{type(error).__name__}: {error}"
    record(1, "a module logger from logging.getLogger(__name__)", ok, detail)

    # 2 --------------------------------------------------------------------
    try:
        collector = Collector()
        logger = logging.getLogger("check.prepare")
        logger.handlers.clear()
        logger.filters.clear()
        logger.setLevel(logging.DEBUG)
        logger.propagate = False
        logger.addHandler(collector)

        kept = mod.prepare(mod.RECORDS, logger)
        levels = [r.levelname for r in collector.records]
        messages = [r.getMessage() for r in collector.records]
        done = [r for r in collector.records if "preparation done" in r.getMessage()]

        ok = (
            len(kept) == 4
            and levels.count("INFO") == 2
            and levels.count("WARNING") == 2
            and levels.count("DEBUG") == 6
            and any("kept 4 of 6" in m for m in messages)
            and SECRET not in " ".join(messages)
            and bool(done)
            and "%" in str(done[0].msg)       # lazy formatting, not an f-string
            and bool(done[0].args)
        )
        detail = f"returned {len(kept)} records; levels {levels}"
    except Exception as error:  # noqa: BLE001
        ok, detail = False, f"{type(error).__name__}: {error}"
    record(2, "prepare() logs at the right levels and returns its result", ok, detail)

    # 3 --------------------------------------------------------------------
    try:
        stream = io.StringIO()
        handler = logging.StreamHandler(stream)
        handler.setLevel(logging.WARNING)          # deliberately too high
        handler.setFormatter(logging.Formatter("%(levelname)s %(message)s"))
        logger = mod.make_debug_logger("check.two_level", handler)
        logger.debug("a debug line")
        text = stream.getvalue()
        ok = (
            logger.level == logging.DEBUG
            and logger.propagate is False
            and len(logger.handlers) == 1
            and "a debug line" in text
        )
        detail = f"logger={logging.getLevelName(logger.level)} " \
                 f"handler={logging.getLevelName(logger.handlers[0].level)} " \
                 f"output={text.strip()!r}"
    except Exception as error:  # noqa: BLE001
        ok, detail = False, f"{type(error).__name__}: {error}"
    record(3, "the two-level trap: a DEBUG call actually comes out", ok, detail)

    # 4 --------------------------------------------------------------------
    try:
        collector = Collector()
        logger = logging.getLogger("check.exception")
        logger.handlers.clear()
        logger.filters.clear()
        logger.setLevel(logging.DEBUG)
        logger.propagate = False
        logger.addHandler(collector)

        bad = mod.log_parse_failure("sixty-four", logger)
        good = mod.log_parse_failure("64", logger)
        failure = collector.records[0]
        rendered = logging.Formatter("%(message)s").format(failure)
        ok = (
            bad is False
            and good is True
            and len(collector.records) == 1
            and failure.levelname == "ERROR"
            and failure.exc_info is not None
            and "Traceback" in rendered
            and "ValueError" in rendered
        )
        detail = f"exc_info={'present' if failure.exc_info else 'MISSING'}"
    except Exception as error:  # noqa: BLE001
        ok, detail = False, f"{type(error).__name__}: {error}"
    record(4, "a failure logged with exception(), traceback attached", ok, detail)

    # 5 --------------------------------------------------------------------
    try:
        formatter = mod.JsonFormatter({"run_id": "run-1"})
        logger, stream = capture("check.json", formatter)
        logger.info("batch complete", extra={"batch": 2, "kept": 61})
        payload = json.loads(stream.getvalue().strip())
        ok = (
            payload["level"] == "INFO"
            and payload["logger"] == "check.json"
            and payload["event"] == "batch complete"
            and payload["run_id"] == "run-1"
            and payload["batch"] == 2
            and payload["kept"] == 61
            and payload["ts"].endswith("Z")
            and payload["ts"][4] == "-"
            and "args" not in payload
        )
        detail = f"parsed keys {sorted(payload)}"
    except Exception as error:  # noqa: BLE001
        ok, detail = False, f"{type(error).__name__}: {error}"
    record(5, "a JSON formatter with ts, level, logger, event and extras", ok, detail)

    # 6 --------------------------------------------------------------------
    try:
        formatter = mod.JsonFormatter({"run_id": "run-1"})
        logger, stream = capture(
            "check.redact", formatter, filters=[mod.RedactingFilter([SECRET])]
        )
        logger.info("calling upstream with key %s", SECRET)
        logger.info(f"in the message: {SECRET}")
        logger.info("config", extra={"headers": {"Authorization": f"Bearer {SECRET}"}})
        text = stream.getvalue()
        ok = SECRET not in text and text.count(mod.RedactingFilter.PLACEHOLDER) >= 3
        detail = ("the secret appears in the captured log" if SECRET in text
                  else "the secret appears nowhere in the captured log")
    except Exception as error:  # noqa: BLE001
        ok, detail = False, f"{type(error).__name__}: {error}"
    record(6, "a redacting filter: the secret reaches no handler", ok, detail)


# ---------------------------------------------------------------------------
# Exercises 7-12
# ---------------------------------------------------------------------------

TOML_TEXT = 'batch_size = 64\nmodel_name = "small-encoder"\ndry_run = false\n'


def check_config(path: Path) -> None:
    try:
        mod = load(path, "starter_config")
    except Exception as error:  # noqa: BLE001
        for number in range(7, 13):
            record(number, "(module did not import)", False, f"{type(error).__name__}: {error}")
        return

    work = Path(tempfile.mkdtemp(prefix="day097-check-"))
    toml_path = work / "config.toml"
    toml_path.write_text(TOML_TEXT, encoding="utf-8")

    # 7 --------------------------------------------------------------------
    try:
        truthy = all(mod.to_bool(t) is True for t in ["true", "TRUE", " yes ", "1", "on"])
        falsey = all(mod.to_bool(t) is False for t in ["false", "FALSE", "no", "0", "off"])
        refused = 0
        for text in ["maybe", "", "2", "y"]:
            try:
                mod.to_bool(text)
            except ValueError:
                refused += 1
        ok = truthy and falsey and refused == 4
        detail = f"truthy={truthy} falsey={falsey} refused {refused} of 4"
    except Exception as error:  # noqa: BLE001
        ok, detail = False, f"{type(error).__name__}: {error}"
    record(7, "to_bool refuses to believe that 'false' is true", ok, detail)

    # 8 --------------------------------------------------------------------
    try:
        table, name = mod.load_toml(toml_path)
        empty, none_name = mod.load_toml(work / "does-not-exist.toml")
        missing, _ = mod.load_toml(None)
        ok = (
            table["batch_size"] == 64
            and isinstance(table["batch_size"], int)
            and table["dry_run"] is False
            and name == "config.toml"
            and empty == {} and none_name is None
            and missing == {}
        )
        detail = f"read {sorted(table)} from {name!r}"
    except Exception as error:  # noqa: BLE001
        ok, detail = False, f"{type(error).__name__}: {error}"
    record(8, "load_toml reads a TOML file and tolerates a missing one", ok, detail)

    # 9 --------------------------------------------------------------------
    try:
        steps = [
            ([], {}, None, 32),
            ([], {}, toml_path, 64),
            ([], {"APP_BATCH_SIZE": "128"}, toml_path, 128),
            (["--batch-size", "256"], {"APP_BATCH_SIZE": "128"}, toml_path, 256),
        ]
        seen = []
        for argv, environ, cfg, expected in steps:
            config = mod.resolve(mod.SPEC, argv=argv, environ=environ, config_path=cfg)
            seen.append((config["batch_size"], expected))
        ok = all(actual == expected for actual, expected in seen)
        detail = " -> ".join(str(actual) for actual, _ in seen)
    except Exception as error:  # noqa: BLE001
        ok, detail = False, f"{type(error).__name__}: {error}"
    record(9, "four layers: default, file, environment, flag — the flag wins", ok, detail)

    # 10 -------------------------------------------------------------------
    try:
        unset = mod.resolve(mod.SPEC, argv=[], environ={}, config_path=toml_path)
        empty = mod.resolve(
            mod.SPEC, argv=[], environ={"APP_MODEL_NAME": ""}, config_path=toml_path
        )
        given = mod.resolve(
            mod.SPEC, argv=[], environ={"APP_MODEL_NAME": "large"}, config_path=toml_path
        )
        int_empty_refused = False
        try:
            mod.resolve(mod.SPEC, argv=[], environ={"APP_BATCH_SIZE": ""})
        except mod.ConfigError as error:
            int_empty_refused = "APP_BATCH_SIZE" in str(error)

        ok = (
            unset["model_name"] == "small-encoder"
            and unset.source_of("model_name").startswith("file:")
            and empty["model_name"] == ""
            and "empty" in empty.source_of("model_name")
            and given["model_name"] == "large"
            and given.source_of("model_name") == "env:APP_MODEL_NAME"
            and int_empty_refused
        )
        detail = (f"unset -> {unset.source_of('model_name')}; "
                  f"empty -> {empty.source_of('model_name')}")
    except Exception as error:  # noqa: BLE001
        ok, detail = False, f"{type(error).__name__}: {error}"
    record(10, "a missing environment variable and an empty one differ", ok, detail)

    # 11 -------------------------------------------------------------------
    try:
        config = mod.resolve(
            mod.SPEC,
            argv=["--batch-size", "256"],
            environ={"APP_API_KEY": SECRET, "APP_LOG_LEVEL": "DEBUG"},
            config_path=toml_path,
        )
        sources = {name: config.source_of(name) for name in
                   ("log_level", "batch_size", "model_name", "dry_run", "api_key")}
        ok = (
            sources["log_level"] == "env:APP_LOG_LEVEL"
            and sources["batch_size"] == "flag:--batch-size"
            and sources["model_name"] == "file:config.toml"
            and sources["dry_run"] == "file:config.toml"
            and sources["api_key"] == "env:APP_API_KEY"
        )
        detail = str(sources)
    except Exception as error:  # noqa: BLE001
        ok, detail = False, f"{type(error).__name__}: {error}"
    record(11, "every value reports the layer it came from", ok, detail)

    # 12 -------------------------------------------------------------------
    try:
        bad = mod.resolve(
            mod.SPEC,
            argv=["--batch-size", "0", "--log-level", "VERBOSE"],
            environ={"APP_API_KEY": SECRET},
            config_path=toml_path,
        )
        problems = mod.validate(bad, mod.SPEC)
        joined = " | ".join(problems)
        good = mod.resolve(mod.SPEC, argv=["--batch-size", "128"], environ={},
                           config_path=toml_path)
        safe = mod.safe_dict(bad)

        ok = (
            len(problems) == 2
            and any("batch_size" in p and "flag:--batch-size" in p for p in problems)
            and any("log_level" in p and "flag:--log-level" in p for p in problems)
            and SECRET not in joined
            and mod.validate(good, mod.SPEC) == []
            and safe["api_key"] == "***redacted***"
            and safe["batch_size"] == 0
            and SECRET not in json.dumps(safe)
        )
        detail = f"{len(problems)} problems: {joined}"
    except Exception as error:  # noqa: BLE001
        ok, detail = False, f"{type(error).__name__}: {error}"
    record(12, "startup validation names the setting and its provenance", ok, detail)

    for leftover in sorted(work.iterdir()):
        leftover.unlink()
    work.rmdir()


def main() -> int:
    if len(sys.argv) != 3:
        print("usage: check_exercises.py LOGGING_FILE CONFIG_FILE", file=sys.stderr)
        return 2
    logging_file, config_file = Path(sys.argv[1]), Path(sys.argv[2])
    sys.path.insert(0, str(logging_file.resolve().parent))

    check_logging(logging_file)
    check_config(config_file)

    passed = 0
    for number, title, ok, detail in sorted(results):
        if ok:
            passed += 1
            print(f"  {number:>2}. ok       {title}")
        else:
            print(f"  {number:>2}. not yet  {title}")
            if detail:
                print(f"          {detail}")
    print()
    print(f"{passed} of 12 exercises complete.")
    return 0 if passed == 12 else 1


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

#!/usr/bin/env bash
# Tests for the Day 097 lab. Run from the lab directory:
#   bash tests/run_tests.sh
#
# Every check compares a REAL VALUE. Log output is captured with a handler
# writing into a buffer — never by scraping stdout — so what is asserted on is
# exactly the records that reached that handler and nothing else.
#
# What this suite asks:
#
#   * does the logging module behave the way the lesson says it does — the
#     two-level trap, propagation and its duplicate, exception() against
#     error(str(e)), lazy formatting?
#   * does the JSON formatter produce parseable objects with the right
#     fields, and does the redacting filter keep the secret out of ALL of
#     them, including the traceback?
#   * does the four-layer resolver put the flag on top, report provenance
#     correctly, tell a missing environment variable from an empty one, and
#     refuse a bad value at startup with a message naming the setting?
#   * does the starter honestly report 0 of 12 before you start, and does the
#     checker say 12 of 12 for the reference answers?
#
# Nothing touches the network. Nothing needs sudo. Everything is built in a
# temporary directory removed by a trap, so a completed run leaves the lab
# directory exactly as it found it.
set -u

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

cleanup() { [ -n "${work}" ] && [ -d "${work}" ] && rm -rf "${work}"; }
trap cleanup EXIT INT TERM

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

# check_eq LABEL EXPECTED ACTUAL — prints what it wanted when it does not match.
check_eq() {
  local label="$1" expected="$2" actual="$3"
  checks=$((checks + 1))
  if [ "${expected}" = "${actual}" ]; then
    echo "  ok: ${label}"
  else
    echo "  FAIL: ${label}"
    echo "        expected: ${expected}"
    echo "        actual:   ${actual}"
    failures=$((failures + 1))
  fi
}

python_bin="${PYTHON:-$(command -v python3 || true)}"
if [ -z "${python_bin}" ] || [ ! -x "${python_bin}" ]; then
  echo "python3 was not found. Install Python 3.11+ or set PYTHON=/path/to/python3."
  exit 1
fi

export PYTHONDONTWRITEBYTECODE=1
work="$(mktemp -d)"

echo "Day 097 — Logging and Configuration"
echo "python3: $("${python_bin}" -c 'import sys; print(sys.version.split()[0])')"
echo "work:    a temporary directory, removed when this script exits"
echo

# tomllib arrived in Python 3.11. Fail with one clear line rather than a wall
# of import errors halfway through.
"${python_bin}" -c 'import tomllib' >/dev/null 2>&1
check "this python has tomllib (3.11 or newer)" \
  "$([ $? -eq 0 ] && echo yes || echo no)"

# ---------------------------------------------------------------------------
# The Python half of the suite. It writes one KEY=VALUE line per assertion
# into a file, which bash then compares. Log capture happens through buffer
# handlers inside this program.
# ---------------------------------------------------------------------------
"${python_bin}" - "${lab_dir}" > "${work}/facts.txt" 2>"${work}/facts.err" <<'PY'
import importlib.util, io, json, logging, os, subprocess, sys, tempfile
from pathlib import Path

lab = Path(sys.argv[1])
sys.path.insert(0, str(lab / "examples"))

from applog import JsonFormatter, RedactingFilter, buffer_handler, iso_utc
import appconfig
from appconfig import APP_SPEC, ConfigError, Setting, resolve, validate, validate_or_die

SECRET = "sk-live-9f2c4a7b1e63"
out = {}

def emit(key, value):
    out[key] = value

# --- the two-level trap ----------------------------------------------------
log = logging.getLogger("t.trap")
log.handlers.clear(); log.filters.clear(); log.propagate = False
log.setLevel(logging.DEBUG)
handler, stream = buffer_handler(logging.WARNING, logging.Formatter("%(message)s"))
log.addHandler(handler)
log.debug("dropped"); log.info("dropped"); log.warning("kept")
emit("TRAP_LINES", len(stream.getvalue().strip().splitlines()))
emit("TRAP_TEXT", stream.getvalue().strip())

log.handlers.clear()
handler, stream = buffer_handler(logging.DEBUG, logging.Formatter("%(message)s"))
log.addHandler(handler)
log.debug("now it comes out")
emit("TRAP_FIXED", stream.getvalue().strip())

# --- propagation and the duplicate ----------------------------------------
root = logging.getLogger()
saved = (root.handlers[:], root.level)
root.handlers.clear(); root.setLevel(logging.DEBUG)
rh, rstream = buffer_handler(logging.DEBUG, logging.Formatter("R:%(message)s"))
root.addHandler(rh)
app = logging.getLogger("t.app")
app.handlers.clear(); app.filters.clear(); app.setLevel(logging.DEBUG); app.propagate = True
ah, astream = buffer_handler(logging.DEBUG, logging.Formatter("A:%(message)s"))
app.addHandler(ah)
child = logging.getLogger("t.app.child")
child.handlers.clear(); child.filters.clear(); child.propagate = True
child.info("once")
emit("PROP_TOTAL", len(astream.getvalue().splitlines()) + len(rstream.getvalue().splitlines()))

app.propagate = False
astream.truncate(0); astream.seek(0); rstream.truncate(0); rstream.seek(0)
child.info("once")
emit("PROP_FIX1", len(astream.getvalue().splitlines()) + len(rstream.getvalue().splitlines()))

app.propagate = True; app.handlers.clear()
astream.truncate(0); astream.seek(0); rstream.truncate(0); rstream.seek(0)
child.info("once")
emit("PROP_FIX2", len(astream.getvalue().splitlines()) + len(rstream.getvalue().splitlines()))
root.handlers.clear(); root.handlers.extend(saved[0]); root.setLevel(saved[1])

# --- exception() against error(str(e)) -------------------------------------
log = logging.getLogger("t.exc")
log.handlers.clear(); log.filters.clear(); log.propagate = False; log.setLevel(logging.DEBUG)
h, s = buffer_handler(logging.DEBUG, logging.Formatter("%(message)s"))
log.addHandler(h)
try:
    int("sixty-four")
except ValueError as error:
    log.error("could not parse: %s", str(error))
emit("ERRSTR_HAS_TRACEBACK", "Traceback" in s.getvalue())
emit("ERRSTR_LINES", len(s.getvalue().strip().splitlines()))
s.truncate(0); s.seek(0)
try:
    int("sixty-four")
except ValueError:
    log.exception("could not parse")
emit("EXC_HAS_TRACEBACK", "Traceback" in s.getvalue())
emit("EXC_NAMES_TYPE", "ValueError" in s.getvalue())
emit("EXC_NAMES_FUNCTION", "int(" in s.getvalue())

# --- lazy formatting -------------------------------------------------------
class Counter:
    n = 0
    def __str__(self):
        Counter.n += 1
        return "rendered"

log = logging.getLogger("t.lazy")
log.handlers.clear(); log.filters.clear(); log.propagate = False
log.setLevel(logging.INFO)
h, s = buffer_handler(logging.INFO, logging.Formatter("%(message)s"))
log.addHandler(h)
Counter.n = 0
for _ in range(100):
    log.debug("x %s", Counter())
emit("LAZY_RENDERS", Counter.n)
Counter.n = 0
for _ in range(100):
    log.debug(f"x {Counter()}")
emit("EAGER_RENDERS", Counter.n)

# --- the JSON formatter ----------------------------------------------------
log = logging.getLogger("t.json")
log.handlers.clear(); log.filters.clear(); log.propagate = False; log.setLevel(logging.DEBUG)
h, s = buffer_handler(logging.DEBUG, JsonFormatter({"run_id": "run-4711"}))
log.addHandler(h)
log.info("batch complete", extra={"batch": 2, "kept": 61})
log.warning("upstream slow", extra={"status": 429})
try:
    int("nope")
except ValueError:
    log.exception("parse failed", extra={"batch": 3})
records = [json.loads(line) for line in s.getvalue().splitlines()]
emit("JSON_COUNT", len(records))
emit("JSON_FIRST_EVENT", records[0]["event"])
emit("JSON_FIRST_LEVEL", records[0]["level"])
emit("JSON_FIRST_RUNID", records[0]["run_id"])
emit("JSON_FIRST_KEPT", records[0]["kept"])
emit("JSON_LOGGER", records[0]["logger"])
emit("JSON_EXC_TYPE", records[2]["exc_type"])
emit("JSON_HAS_TRACEBACK_FIELD", "traceback" in records[2])
emit("JSON_TS_SHAPE", len(records[0]["ts"]) == 24 and records[0]["ts"].endswith("Z"))
emit("JSON_TS_SORTS", sorted(r["ts"] for r in records) == [r["ts"] for r in records])
emit("ISO_UTC_ZERO", iso_utc(0))

# --- the redacting filter --------------------------------------------------
def redacted_text(on_logger):
    logger = logging.getLogger("t.redact.logger" if on_logger else "t.redact.handler")
    logger.handlers.clear(); logger.filters.clear()
    logger.propagate = False; logger.setLevel(logging.DEBUG)
    h, s = buffer_handler(logging.DEBUG, JsonFormatter({"run_id": "run-4711"}))
    if on_logger:
        logger.addFilter(RedactingFilter([SECRET]))
    else:
        h.addFilter(RedactingFilter([SECRET]))
    logger.addHandler(h)
    logger.info("key %s", SECRET)
    logger.info(f"inline {SECRET}")
    logger.info("cfg", extra={"headers": {"Authorization": f"Bearer {SECRET}"}})
    logger.info("list", extra={"argv": ["--key", SECRET]})
    return logger, s

logger, s = redacted_text(on_logger=False)
emit("REDACT_SECRET_PRESENT", SECRET in s.getvalue())
emit("REDACT_PLACEHOLDERS", s.getvalue().count("***redacted***"))

# the same filter attached to the LOGGER, and a record arriving from a child
logger, s = redacted_text(on_logger=True)
child = logging.getLogger("t.redact.logger.child")
child.handlers.clear(); child.filters.clear(); child.propagate = True
child.info("key %s", SECRET)
lines = s.getvalue().splitlines()
emit("LOGGER_FILTER_DIRECT_LEAKS", SECRET in lines[0])
emit("LOGGER_FILTER_CHILD_LEAKS", SECRET in lines[-1])

# a secret inside an exception message survives a filter, and does not
# survive a formatter that scrubs the finished line
class ScrubbingJsonFormatter(JsonFormatter):
    def __init__(self, secrets, **kwargs):
        super().__init__(**kwargs)
        self.redactor = RedactingFilter(secrets)
    def format(self, record):
        return self.redactor.scrub(super().format(record))

for label, formatter in (("FILTERONLY", JsonFormatter()),
                         ("SCRUBBED", ScrubbingJsonFormatter([SECRET]))):
    logger = logging.getLogger(f"t.exc.{label}")
    logger.handlers.clear(); logger.filters.clear()
    logger.propagate = False; logger.setLevel(logging.DEBUG)
    h, s = buffer_handler(logging.DEBUG, formatter)
    h.addFilter(RedactingFilter([SECRET]))
    logger.addHandler(h)
    try:
        raise RuntimeError(f"upstream rejected key {SECRET}")
    except RuntimeError:
        logger.exception("request failed")
    emit(f"TRACEBACK_{label}_LEAKS", SECRET in s.getvalue())

# --- configuration: four layers -------------------------------------------
cfg = lab / "examples" / "config.toml"
steps = [
    ([], {}, None),
    ([], {}, cfg),
    ([], {"APP_BATCH_SIZE": "128"}, cfg),
    (["--batch-size", "256"], {"APP_BATCH_SIZE": "128"}, cfg),
]
values, sources = [], []
for argv, environ, path in steps:
    c = resolve(APP_SPEC, argv=argv, environ=environ, config_path=path)
    values.append(c["batch_size"])
    sources.append(c.source_of("batch_size"))
emit("LAYER_VALUES", ",".join(str(v) for v in values))
emit("LAYER_SOURCES", ",".join(sources))

c = resolve(APP_SPEC, argv=["--batch-size", "256", "--log-level", "DEBUG"],
            environ={"APP_SEED": "7", "APP_API_KEY": SECRET}, config_path=cfg)
emit("PROV_TABLE", ",".join(f"{n}={r.source}" for n, r in c.settings.items()))
emit("PROV_TABLE_LEAKS", SECRET in c.provenance_table())
emit("SAFE_DICT_LEAKS", SECRET in json.dumps(c.safe_dict()))
emit("SAFE_DICT_KEY", c.safe_dict()["api_key"])
emit("AS_DICT_HAS_SECRET", c.as_dict()["api_key"] == SECRET)

# TOML types survive
c = resolve(APP_SPEC, argv=[], environ={}, config_path=cfg)
emit("TOML_TYPES", f"{type(c['batch_size']).__name__},{type(c['dry_run']).__name__}")

# --- missing against empty -------------------------------------------------
unset = resolve(APP_SPEC, argv=[], environ={}, config_path=cfg)
empty = resolve(APP_SPEC, argv=[], environ={"APP_MODEL_NAME": ""}, config_path=cfg)
given = resolve(APP_SPEC, argv=[], environ={"APP_MODEL_NAME": "large"}, config_path=cfg)
emit("EMPTY_TRIPLE", "|".join([
    f"{unset['model_name']}:{unset.source_of('model_name')}",
    f"{empty['model_name']}:{empty.source_of('model_name')}",
    f"{given['model_name']}:{given.source_of('model_name')}",
]))
try:
    resolve(APP_SPEC, argv=[], environ={"APP_BATCH_SIZE": ""})
    emit("EMPTY_INT_REFUSED", False)
except ConfigError as error:
    emit("EMPTY_INT_REFUSED", "APP_BATCH_SIZE" in str(error))

# --- bool conversion -------------------------------------------------------
emit("BOOL_NAIVE", bool("false"))
emit("BOOL_TRUE_WORDS", ",".join(str(appconfig.to_bool(t)) for t in
                                 ["true", "TRUE", " yes ", "1", "on"]))
emit("BOOL_FALSE_WORDS", ",".join(str(appconfig.to_bool(t)) for t in
                                  ["false", "FALSE", "no", "0", "off"]))
refused = 0
for text in ["maybe", "", "2", "y"]:
    try:
        appconfig.to_bool(text)
    except ValueError:
        refused += 1
emit("BOOL_REFUSED", refused)
c = resolve(APP_SPEC, argv=[], environ={"APP_DRY_RUN": "false"}, config_path=cfg)
emit("BOOL_FROM_ENV", c["dry_run"])

# --- validation ------------------------------------------------------------
bad = resolve(APP_SPEC, argv=["--batch-size", "0", "--log-level", "VERBOSE"],
              environ={"APP_SEED": "-1", "APP_API_KEY": SECRET}, config_path=cfg)
problems = validate(bad, APP_SPEC)
emit("VALIDATE_COUNT", len(problems))
emit("VALIDATE_NAMES_SETTING", all(p.split(":")[0] in
     {"log_level", "batch_size", "seed"} for p in problems))
emit("VALIDATE_NAMES_SOURCE", all(("flag:" in p or "env:" in p or "file:" in p
     or "default" in p) for p in problems))
emit("VALIDATE_LEAKS", SECRET in " ".join(problems))
good = resolve(APP_SPEC, argv=["--batch-size", "128"], environ={"APP_SEED": "7"},
               config_path=cfg)
emit("VALIDATE_GOOD", len(validate(good, APP_SPEC)))
try:
    validate_or_die(bad, APP_SPEC)
    emit("VALIDATE_DIED", False)
except ConfigError:
    emit("VALIDATE_DIED", True)

# --- 06_run_manifest.py, end to end ---------------------------------------
env = dict(os.environ)
env.update({"APP_API_KEY": SECRET, "APP_SEED": "7", "PYTHONDONTWRITEBYTECODE": "1"})
proc = subprocess.run(
    [sys.executable, str(lab / "examples" / "06_run_manifest.py")],
    capture_output=True, text=True, env=env, cwd=str(lab),
)
emit("MANIFEST_EXIT", proc.returncode)
lines = [line for line in proc.stdout.splitlines() if line.startswith("{")]
manifest = json.loads(lines[0])
emit("MANIFEST_EVENT", manifest["event"])
emit("MANIFEST_RUNID", manifest["run_id"])
emit("MANIFEST_SEED_SOURCE", manifest["provenance"]["seed"])
emit("MANIFEST_BATCH_SOURCE", manifest["provenance"]["batch_size"])
emit("MANIFEST_KEY", manifest["config"]["api_key"])
emit("MANIFEST_LEAKS", SECRET in proc.stdout or SECRET in proc.stderr)
emit("MANIFEST_LINES", len(lines))
emit("MANIFEST_FINAL_LOSS", json.loads(lines[-1])["final_loss"])
emit("MANIFEST_RUNIDS_ALL", len({json.loads(l)["run_id"] for l in lines}))

proc2 = subprocess.run(
    [sys.executable, str(lab / "examples" / "06_run_manifest.py"),
     "--seed", "7", "--batch-size", "64", "--model-name", "small-encoder",
     "--data-version", "2026-08-01"],
    capture_output=True, text=True, env=env, cwd=str(lab),
)
lines2 = [line for line in proc2.stdout.splitlines() if line.startswith("{")]
emit("MANIFEST_REPRODUCED",
     json.loads(lines2[-1])["final_loss"] == json.loads(lines[-1])["final_loss"])

proc3 = subprocess.run(
    [sys.executable, str(lab / "examples" / "06_run_manifest.py"), "--batch-size", "0"],
    capture_output=True, text=True, env=env, cwd=str(lab),
)
emit("MANIFEST_BAD_EXIT", proc3.returncode)
emit("MANIFEST_BAD_NAMES_SETTING", "batch_size" in proc3.stderr)
emit("MANIFEST_BAD_NAMES_SOURCE", "flag:--batch-size" in proc3.stderr)

for key, value in out.items():
    print(f"{key}={value}")
PY

if [ -s "${work}/facts.err" ]; then
  echo "the python half of the suite failed:"
  cat "${work}/facts.err"
  exit 1
fi

fact() { grep "^$1=" "${work}/facts.txt" | head -1 | cut -d= -f2-; }

# ---------------------------------------------------------------------------
echo
echo "1. The logging module behaves as the lesson claims"
# ---------------------------------------------------------------------------
check_eq "the two-level trap: 3 calls, logger at DEBUG, handler at WARNING -> 1 line" \
  "1" "$(fact TRAP_LINES)"
check_eq "and the line that survived is the warning" "kept" "$(fact TRAP_TEXT)"
check_eq "lowering the HANDLER's level lets the debug line out" \
  "now it comes out" "$(fact TRAP_FIXED)"
check_eq "propagation: one call, two handlers up the tree -> 2 lines" \
  "2" "$(fact PROP_TOTAL)"
check_eq "fix 1, propagate = False -> 1 line" "1" "$(fact PROP_FIX1)"
check_eq "fix 2, handlers in one place only -> 1 line" "1" "$(fact PROP_FIX2)"

# ---------------------------------------------------------------------------
echo
echo "2. exception() keeps what error(str(e)) throws away"
# ---------------------------------------------------------------------------
check_eq "log.error(str(e)) produces no traceback" "False" "$(fact ERRSTR_HAS_TRACEBACK)"
check_eq "log.error(str(e)) is one line" "1" "$(fact ERRSTR_LINES)"
check_eq "log.exception() attaches the traceback" "True" "$(fact EXC_HAS_TRACEBACK)"
check_eq "the traceback names the exception type" "True" "$(fact EXC_NAMES_TYPE)"
check_eq "the traceback names the failing call" "True" "$(fact EXC_NAMES_FUNCTION)"

# ---------------------------------------------------------------------------
echo
echo "3. Lazy formatting renders nothing for a suppressed record"
# ---------------------------------------------------------------------------
check_eq "100 suppressed DEBUG calls, %s formatting -> 0 renders" \
  "0" "$(fact LAZY_RENDERS)"
check_eq "100 suppressed DEBUG calls, f-string -> 100 renders" \
  "100" "$(fact EAGER_RENDERS)"

# ---------------------------------------------------------------------------
echo
echo "4. The JSON formatter produces parseable objects with real fields"
# ---------------------------------------------------------------------------
check_eq "three calls produced three JSON objects" "3" "$(fact JSON_COUNT)"
check_eq "event is the formatted message" "batch complete" "$(fact JSON_FIRST_EVENT)"
check_eq "level is the NAME, not the number" "INFO" "$(fact JSON_FIRST_LEVEL)"
check_eq "the logger's name is carried" "t.json" "$(fact JSON_LOGGER)"
check_eq "run_id is stamped on every line by the formatter" \
  "run-4711" "$(fact JSON_FIRST_RUNID)"
check_eq "a field passed through extra= survives as a field" "61" "$(fact JSON_FIRST_KEPT)"
check_eq "the exception's type is its own field" "ValueError" "$(fact JSON_EXC_TYPE)"
check_eq "the traceback is its own field, not glued to the message" \
  "True" "$(fact JSON_HAS_TRACEBACK_FIELD)"
check_eq "ts is ISO 8601 UTC to milliseconds" "True" "$(fact JSON_TS_SHAPE)"
check_eq "ts sorts chronologically as plain text" "True" "$(fact JSON_TS_SORTS)"
check_eq "iso_utc(0) is the Unix epoch in UTC" \
  "1970-01-01T00:00:00.000Z" "$(fact ISO_UTC_ZERO)"

# ---------------------------------------------------------------------------
echo
echo "5. The secret does not reach the log"
# ---------------------------------------------------------------------------
check_eq "with the filter on the handler, the secret appears NOWHERE" \
  "False" "$(fact REDACT_SECRET_PRESENT)"
check_eq "and four routes were redacted: message, args, nested dict, list" \
  "4" "$(fact REDACT_PLACEHOLDERS)"
check_eq "a filter on the LOGGER protects a direct call" \
  "False" "$(fact LOGGER_FILTER_DIRECT_LEAKS)"
check_eq "and DOES NOT protect a record propagating from a child logger" \
  "True" "$(fact LOGGER_FILTER_CHILD_LEAKS)"
check_eq "a secret inside an exception message survives a filter" \
  "True" "$(fact TRACEBACK_FILTERONLY_LEAKS)"
check_eq "and does not survive a formatter that scrubs the finished line" \
  "False" "$(fact TRACEBACK_SCRUBBED_LEAKS)"

# ---------------------------------------------------------------------------
echo
echo "6. Configuration: four layers, in order, each one overriding the last"
# ---------------------------------------------------------------------------
check_eq "default 32, file 64, environment 128, flag 256" \
  "32,64,128,256" "$(fact LAYER_VALUES)"
check_eq "and each value reports the layer it came from" \
  "default,file:config.toml,env:APP_BATCH_SIZE,flag:--batch-size" \
  "$(fact LAYER_SOURCES)"
check_eq "seven settings, five different provenances" \
  "log_level=flag:--log-level,batch_size=flag:--batch-size,model_name=file:config.toml,seed=env:APP_SEED,dry_run=file:config.toml,data_version=file:config.toml,api_key=env:APP_API_KEY" \
  "$(fact PROV_TABLE)"
check_eq "TOML has real types: no conversion needed for int or bool" \
  "int,bool" "$(fact TOML_TYPES)"

# ---------------------------------------------------------------------------
echo
echo "7. Missing and empty are different, and strings are not types"
# ---------------------------------------------------------------------------
check_eq "unset falls through; empty is a value; set is a value" \
  "small-encoder:file:config.toml|:env:APP_MODEL_NAME (set but empty)|large:env:APP_MODEL_NAME" \
  "$(fact EMPTY_TRIPLE)"
check_eq "an empty environment variable for an int setting is refused by name" \
  "True" "$(fact EMPTY_INT_REFUSED)"
check_eq "the trap: bool('false') is True" "True" "$(fact BOOL_NAIVE)"
check_eq "to_bool reads the true words" "True,True,True,True,True" \
  "$(fact BOOL_TRUE_WORDS)"
check_eq "to_bool reads the false words" "False,False,False,False,False" \
  "$(fact BOOL_FALSE_WORDS)"
check_eq "to_bool refuses all four ambiguous inputs" "4" "$(fact BOOL_REFUSED)"
check_eq "APP_DRY_RUN=false resolves to False, not True" "False" "$(fact BOOL_FROM_ENV)"

# ---------------------------------------------------------------------------
echo
echo "8. Startup validation, and the secret it must not print"
# ---------------------------------------------------------------------------
check_eq "three bad values are reported all at once" "3" "$(fact VALIDATE_COUNT)"
check_eq "every message names its setting" "True" "$(fact VALIDATE_NAMES_SETTING)"
check_eq "every message names the layer the value came from" \
  "True" "$(fact VALIDATE_NAMES_SOURCE)"
check_eq "no message contains the secret" "False" "$(fact VALIDATE_LEAKS)"
check_eq "a good configuration reports no problems" "0" "$(fact VALIDATE_GOOD)"
check_eq "validate_or_die raises on a bad configuration" "True" "$(fact VALIDATE_DIED)"
check_eq "the provenance table never prints the secret" "False" "$(fact PROV_TABLE_LEAKS)"
check_eq "safe_dict never contains the secret" "False" "$(fact SAFE_DICT_LEAKS)"
check_eq "safe_dict shows the placeholder instead" "***redacted***" "$(fact SAFE_DICT_KEY)"
check_eq "as_dict DOES carry the real value, because the program needs it" \
  "True" "$(fact AS_DICT_HAS_SECRET)"

# ---------------------------------------------------------------------------
echo
echo "9. The run manifest: a run reconstructable from its own log"
# ---------------------------------------------------------------------------
check_eq "06_run_manifest.py exits 0" "0" "$(fact MANIFEST_EXIT)"
check_eq "its first event is the manifest" "run started" "$(fact MANIFEST_EVENT)"
check_eq "every line carries one run_id" "1" "$(fact MANIFEST_RUNIDS_ALL)"
check_eq "the run id is on the manifest line" "run-4711" "$(fact MANIFEST_RUNID)"
check_eq "the manifest records where the seed came from" \
  "env:APP_SEED" "$(fact MANIFEST_SEED_SOURCE)"
check_eq "and where the batch size came from" \
  "file:config.toml" "$(fact MANIFEST_BATCH_SOURCE)"
check_eq "the manifest carries the placeholder, not the key" \
  "***redacted***" "$(fact MANIFEST_KEY)"
check_eq "the key appears in neither stdout nor stderr" "False" "$(fact MANIFEST_LEAKS)"
check_eq "six JSON events were logged" "6" "$(fact MANIFEST_LINES)"
check_eq "the final loss is deterministic for seed 7" \
  "0.506509" "$(fact MANIFEST_FINAL_LOSS)"
check_eq "re-running from the manifest reproduces the same final loss" \
  "True" "$(fact MANIFEST_REPRODUCED)"
check_eq "a bad configuration stops the program before it starts" \
  "2" "$(fact MANIFEST_BAD_EXIT)"
check_eq "and the refusal names the setting" "True" "$(fact MANIFEST_BAD_NAMES_SETTING)"
check_eq "and names the layer it came from" "True" "$(fact MANIFEST_BAD_NAMES_SOURCE)"

# ---------------------------------------------------------------------------
echo
echo "10. The demonstration scripts and the starter checker"
# ---------------------------------------------------------------------------
for script in 01_prints.py 02_logging_architecture.py 03_structured_logging.py \
              04_config_resolver.py 05_dictconfig_and_rotation.py; do
  (cd "${lab_dir}" && "${python_bin}" "examples/${script}" >"${work}/${script}.out" 2>&1)
  check "examples/${script} runs and exits 0" \
    "$([ $? -eq 0 ] && echo yes || echo no)"
done
check "05_dictconfig_and_rotation.py rotated the file into 4 generations" \
  "$(grep -q 'app.log.3' "${work}/05_dictconfig_and_rotation.py.out" && echo yes || echo no)"
check "05 shows a TimedRotatingFileHandler rollover" \
  "$(grep -q 'files after one rollover: 2' "${work}/05_dictconfig_and_rotation.py.out" \
     && echo yes || echo no)"
check "01_prints.py really does print the API key, which is the point" \
  "$(grep -q 'sk-live-9f2c4a7b1e63' "${work}/01_prints.py.out" && echo yes || echo no)"
check "no demonstration script leaves an absolute home path in its output" \
  "$(grep -l '/Users/\|/home/' "${work}"/*.out >/dev/null 2>&1 && echo no || echo yes)"

(cd "${lab_dir}" && bash starter/03_check.sh >"${work}/starter-before.txt" 2>&1)
before_status=$?
check_eq "the untouched starter reports 0 of 12" "0 of 12 exercises complete." \
  "$(grep 'exercises complete' "${work}/starter-before.txt")"
check_eq "and exits non-zero" "1" "${before_status}"

(cd "${lab_dir}" && bash starter/03_check.sh examples/07_solution_logging.py \
   examples/08_solution_config.py >"${work}/starter-after.txt" 2>&1)
after_status=$?
check_eq "the reference answers report 12 of 12" "12 of 12 exercises complete." \
  "$(grep 'exercises complete' "${work}/starter-after.txt")"
check_eq "and exit 0" "0" "${after_status}"

# The checker must be able to FAIL. Break one reference answer on purpose and
# confirm the count drops — a checker that always says 12 is worth nothing.
cp "${lab_dir}/examples/applog.py" "${lab_dir}/examples/appconfig.py" "${work}/"
sed 's/logger.exception("could not parse batch size")/logger.error("could not parse batch size")/' \
  "${lab_dir}/examples/07_solution_logging.py" > "${work}/broken_logging.py"
(cd "${lab_dir}" && bash starter/03_check.sh "${work}/broken_logging.py" \
   examples/08_solution_config.py >"${work}/starter-broken.txt" 2>&1)
check_eq "with exception() replaced by error(), the checker catches it" \
  "11 of 12 exercises complete." \
  "$(grep 'exercises complete' "${work}/starter-broken.txt")"

# ---------------------------------------------------------------------------
echo
echo "11. Hygiene: offline, no sudo, no leaked paths, nothing left behind"
# ---------------------------------------------------------------------------
"${python_bin}" - "${lab_dir}" > "${work}/hygiene.txt" <<'PY'
import re, sys
from pathlib import Path

root = Path(sys.argv[1])
urls, sudo_lines = set(), []
comment = re.compile(r"^\s*#")
for path in sorted(root.rglob("*")):
    if not path.is_file() or path.suffix not in {".py", ".sh", ".toml"}:
        continue
    for number, line in enumerate(
        path.read_text(encoding="utf-8", errors="ignore").splitlines(), 1
    ):
        urls.update(re.findall(r"https?://[^\s\"')]+", line))
        if re.search(r"(^|[;|&(]\s*)sudo\s", line) and not comment.match(line):
            sudo_lines.append(f"{path.name}:{number}")
print("URLS " + " ".join(sorted(urls)))
print("SUDO " + " ".join(sudo_lines))
PY
check_eq "no URL appears anywhere in the lab's scripts" "URLS" \
  "$(grep '^URLS ' "${work}/hygiene.txt" | sed 's/ *$//')"
check_eq "no line in this lab would actually invoke sudo" "SUDO" \
  "$(grep '^SUDO ' "${work}/hygiene.txt" | sed 's/ *$//')"
check "nothing in this lab imports a networking module" \
  "$(grep -rlE '^\s*(import|from)\s+(socket|urllib|http|requests)' \
     "${lab_dir}/examples" "${lab_dir}/starter" >/dev/null 2>&1 && echo no || echo yes)"
check "no captured output leaks an absolute home path" \
  "$(grep -rl '/Users/\|/home/' "${lab_dir}/expected-output" >/dev/null 2>&1 \
     && echo no || echo yes)"
check "this suite left no log file in the lab directory" \
  "$([ ! -e "${lab_dir}/app.log" ] && [ ! -e "${lab_dir}/daily.log" ] \
     && [ ! -e "${lab_dir}/starter/app.log" ] && echo yes || echo no)"
check "this suite left no __pycache__ behind" \
  "$(find "${lab_dir}" -type d -name __pycache__ | grep -q . && echo no || echo yes)"

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