Part 11 · Module 1: Engineering Floor

Lab: The claims service skeleton, end to end

18 min read·13 Oct 2026

Module 1 · Lab · About 75 minutes · Needs: lessons 1 to 4 (Git and tests, the FastAPI backend, the screen's rows, config and CI)

What you are building

The finished slice: Dana's email is recorded once, triaged in the background by the model (amber), and served to the screen (blue), with tests, CI, Compose and secret hygiene around it (green).

Module 1 adds the engagement's project skeleton: tests, CI, Docker Compose and secrets handled properly. This lab joins the four lessons' parts into one repository, harbourline-claims, that works as a slice: Dana Price's email, MSG-3101, goes into an API, is recorded once, waits in a queue, is triaged by the model, and the open exceptions come back in the shape Lesson 3's screen draws. Every piece has a test, the tests run without a model key, the service refuses to start without its settings, and no secret reaches Git. Everything in it is simulated.

Step 1: Lay out the repository

Rules first: .gitignore keeps .env, compiled files and the local database out of Git (red), and .env.example records the names of the settings without their values (green).

A repository Mei Lin can inherit starts with its rules. Decide what Git must never see (.env, compiled files, the local database) and list the settings in .env.example, names without values. The shared data goes in as a seed file, so the service and its tests start from the same rows as every lesson.

python

%pip install -q -U langchain langgraph langchain-groq langchain-google-genai langchain-ollama

python

import os
from langchain.chat_models import init_chat_model

PROVIDER = "groq"   # "groq" or "gemini" in Colab; "ollama" on your own computer
MODELS = {"groq": "groq:openai/gpt-oss-20b", "gemini": "google_genai:gemini-3.5-flash-lite", "ollama": "ollama:llama3.2:3b"}
KEY = {"groq": "GROQ_API_KEY", "gemini": "GOOGLE_API_KEY", "ollama": None}[PROVIDER]
if KEY and KEY not in os.environ:
    from google.colab import userdata   # keys live in Colab's Secrets panel
    os.environ[KEY] = userdata.get(KEY)
llm = init_chat_model(MODELS[PROVIDER], temperature=0, max_retries=6)

Code walkthrough The course's standard setup cell; max_retries=6 waits out rate limits. Only step 6 uses llm; nothing else needs a key.

python

%pip install -q fastapi httpx fakeredis pytest

python

from datetime import date

TMS_CUSTOMERS = {
    "C-1042": "NORTHGATE BUILDING SUPPLIES LTD",
    "C-1077": "Pinder & Sons Hardware",
    "C-1130": "GREENWAY GARDEN CTR",
    "C-1188": "Atlas Office Furniture",
    "C-1203": "ST. MARY'S HOSPITAL TRUST",
}
SHIPMENTS = {
    "HF-88101": {"customer": "C-1042", "promised": "2026-03-02", "delivered": "2026-03-02", "freight_charge": 180.00, "declared_value": None, "weight_kg": 420},
    "HF-88102": {"customer": "C-1042", "promised": "2026-03-03", "delivered": "2026-03-09", "freight_charge": 240.00, "declared_value": None, "weight_kg": 610},
    "HF-88145": {"customer": "C-1077", "promised": "2026-03-04", "delivered": "2026-03-04", "freight_charge": 95.00, "declared_value": None, "weight_kg": 60},
    "HF-88190": {"customer": "C-1130", "promised": "2026-03-05", "delivered": None, "freight_charge": 130.00, "declared_value": None, "weight_kg": 150},
    "HF-88207": {"customer": "C-1188", "promised": "2026-03-06", "delivered": "2026-03-06", "freight_charge": 310.00, "declared_value": 6500.00, "weight_kg": 380},
    "HF-88233": {"customer": "C-1203", "promised": "2026-03-09", "delivered": "2026-03-10", "freight_charge": 75.00, "declared_value": None, "weight_kg": 25},
    "HF-88260": {"customer": "C-1077", "promised": "2026-03-10", "delivered": "2026-03-10", "freight_charge": 110.00, "declared_value": None, "weight_kg": 90},
    "HF-88301": {"customer": "C-1188", "promised": "2026-03-11", "delivered": "2026-03-16", "freight_charge": 420.00, "declared_value": None, "weight_kg": 700},
}
print(len(TMS_CUSTOMERS), "TMS customers,", len(SHIPMENTS), "shipments")

Code walkthrough The TMS customers and the eight shipments, copied exactly from the shared data. It prints 5 TMS customers, 8 shipments.

python

EXCEPTIONS_MASTER = [
    {"ref": "HF-88102", "type": "late", "opened": "2026-03-09", "owner": "Joan", "colour": "amber", "notes": "gold acct - Rafael says credit without arguing", "closed": "2026-03-11"},
    {"ref": "HF-88207", "type": "damage", "opened": "2026-03-09", "owner": "Dev", "colour": "red", "notes": "DV 6500 so above cap, needs Tom. waiting on photos", "closed": None},
    {"ref": "HF-88190", "type": "missing", "opened": "2026-03-13", "owner": "Joan", "colour": "red", "notes": "depot says on trailer 14?? chase Tues", "closed": None},
    {"ref": "HF-88233", "type": "late", "opened": "2026-03-10", "owner": "Priti", "colour": "green", "notes": "1 day only, no credit due. hospital - reply same day anyway", "closed": "2026-03-10"},
    {"ref": "HF-88301", "type": "late", "opened": "2026-03-16", "owner": "Dev", "colour": "amber", "notes": "5 days late. 10% = 42.00", "closed": None},
]
INBOX = [
    {"id": "MSG-3101", "from": "dana@northgate.example", "subject": "HF-88102 late again",
     "body": "Hi, HF-88102 was promised for the 3rd and turned up on the 9th. Call me on 07700 900123 if you need anything. Dana Price, Site Manager"},
]
print(len(EXCEPTIONS_MASTER), "spreadsheet rows,", len(INBOX), "email")

Code walkthrough Joan's spreadsheet rows and the first email from the claims inbox, copied exactly. It prints 5 spreadsheet rows, 1 email.

python

import shutil, subprocess, sys, textwrap
from pathlib import Path

REPO = Path("harbourline-claims")
shutil.rmtree(REPO, ignore_errors=True)                       # start clean on a rerun
for folder in ("claims", "tests", ".github/workflows"):
    (REPO / folder).mkdir(parents=True, exist_ok=True)

def write(path, text):
    (REPO / path).write_text(textwrap.dedent(text).lstrip())

def git(*args):
    r = subprocess.run(["git", "-c", "user.name=You", "-c", "user.email=you@larkfield.example", *args],
                       cwd=REPO, capture_output=True, text=True)
    return (r.stdout + r.stderr).strip()

write("claims/__init__.py", "")
write(".gitignore", ".env\n__pycache__/\n*.pyc\n*.db\n")
write(".env.example", "DATABASE_URL=\nREDIS_URL=\nAPI_KEYS=\nMODEL=\nGROQ_API_KEY=\n")
write("claims/seed.py", "\n".join(["from datetime import date", "DATA_AS_OF = date(2026, 3, 16)",
      f"TMS_CUSTOMERS = {TMS_CUSTOMERS!r}", f"SHIPMENTS = {SHIPMENTS!r}",
      f"EXCEPTIONS_MASTER = {EXCEPTIONS_MASTER!r}", f"INBOX = {INBOX!r}", ""]))
git("init", "-b", "main")
print(sorted(str(p.relative_to(REPO)) for p in REPO.rglob("*") if p.is_file() and "/.git/" not in str(p)))

Code walkthrough

REPO is Lesson 1's folder name; shutil.rmtree wipes it so a rerun starts clean. git is Lesson 1's helper.
write saves a file in the repository; textwrap.dedent strips the indentation used in the blocks below.
seed.py is generated with !r, which writes each value as Python can read it back, so the seed matches the shared data exactly.

You should see ['.env.example', '.gitignore', 'claims/__init__.py', 'claims/seed.py'].

Step 2: Pin the late-delivery rule

The rule every later piece calls: HF-88301 is 3 working days late once the weekend (grey) is skipped, so the credit is £42.00, pinned by three passing tests.

The rule the engagement is judged on goes in first, because every later piece calls it. Lesson 1's days_late and Lesson 4's late_credit live together in claims/lateness.py, pinned by three tests: HF-88301, HF-88233 and the Easter weekend.

python

write("claims/lateness.py", '''
    from datetime import date, timedelta

    BANK_HOLIDAYS_2026 = [date(2026, 1, 1), date(2026, 4, 3), date(2026, 4, 6), date(2026, 5, 4),
                          date(2026, 5, 25), date(2026, 8, 31), date(2026, 12, 25), date(2026, 12, 28)]

    def days_late(promised: date, delivered: date) -> int:
        """Working days after `promised`, up to and including `delivered`."""
        days, d = 0, promised
        while d < delivered:
            d += timedelta(days=1)
            if d.weekday() < 5 and d not in BANK_HOLIDAYS_2026:
                days += 1
        return days

    def late_credit(freight_charge: float, working_days_late: int) -> float:
        return round(0.10 * freight_charge, 2) if working_days_late > 2 else 0.0
''')

Code walkthrough

days_late is the shared working-day function: it counts Monday to Friday, skipping the eight bank holidays of England and Wales.
late_credit is the policy: 10% of the freight charge when more than 2 working days late, otherwise nothing.

python

write("tests/test_lateness.py", '''
    from datetime import date
    from claims.lateness import days_late, late_credit

    def test_hf_88301_three_working_days_late_credit_42():
        late = days_late(date(2026, 3, 11), date(2026, 3, 16))
        assert (late, late_credit(420.00, late)) == (3, 42.00)

    def test_hf_88233_one_day_late_no_credit():
        late = days_late(date(2026, 3, 9), date(2026, 3, 10))
        assert (late, late_credit(75.00, late)) == (1, 0.0)

    def test_easter_weekend_is_one_working_day():
        assert days_late(date(2026, 4, 2), date(2026, 4, 7)) == 1
''')

def ci():
    r = subprocess.run([sys.executable, "-B", "-m", "pytest", "-q", "-p", "no:cacheprovider"],
                       cwd=REPO, capture_output=True, text=True)
    return f"exit code {r.returncode}: {r.stdout.strip().splitlines()[-1]}"

print(ci())

Code walkthrough

Each test checks a pair, the days and the credit, so one assert pins both halves of the rule.
ci runs pytest as the pipeline will and returns the exit code with pytest's last line. -B stops Python caching compiled files, so later edits are seen.

You should see exit code 0: 3 passed, then pytest's timing.

Step 3: Settings that fail fast

Four settings are read once at start-up; the secret (red) is never printed, and a missing setting stops the service before anyone uses it.

The service needs four settings: the database, the queue, the API keys and the model. API_KEYS is a secret. Lesson 4's rule holds: read them once at start-up, refuse to start if any is missing, never print a secret.

python

write("claims/config.py", '''
    import os

    REQUIRED = ["DATABASE_URL", "REDIS_URL", "API_KEYS", "MODEL"]
    SECRETS = {"API_KEYS"}

    class ConfigError(Exception):
        pass

    def load_config(env=os.environ):
        missing = [name for name in REQUIRED if not env.get(name)]
        if missing:
            raise ConfigError("missing settings: " + ", ".join(missing))
        config = {name: env[name] for name in REQUIRED}
        config["API_KEYS"] = dict(pair.split(":") for pair in env["API_KEYS"].split(","))
        return config

    def safe_to_log(config):
        return {k: ("****" if k in SECRETS else v) for k, v in config.items()}
''')
write("tests/test_config.py", '''
    import pytest
    from claims.config import ConfigError, load_config

    GOOD = {"DATABASE_URL": "sqlite:///:memory:", "REDIS_URL": "redis://unused", "API_KEYS": "k1:clerk", "MODEL": "none"}

    def test_missing_keys_refuse_to_start():
        with pytest.raises(ConfigError, match="API_KEYS"):
            load_config({**GOOD, "API_KEYS": ""})

    def test_keys_become_roles():
        assert load_config(GOOD)["API_KEYS"] == {"k1": "clerk"}
''')
print(ci())

Code walkthrough

missing collects every absent setting, so one error names them all.
API_KEYS arrives as text such as key-a:clerk,key-b:viewer and becomes a dictionary from key to role, the shape Lesson 2's role_of used.
safe_to_log hides the whole secret with ****.
The tests prove a blank key stops start-up and that keys turn into roles.

You should see exit code 0: 5 passed: the three policy tests and the two new ones.

Step 4: A store that keeps facts once

Lesson 2's three tables, with one addition: a UNIQUE message_id (green) lets the database itself refuse a second record for the same email.

Lesson 2's three tables move into claims/db.py with one change: exceptions gains a message_id column marked UNIQUE. The email becomes the idempotency key, held by the database, so a retry is caught even after a restart. Lesson 2's dictionary of keys was lost on restart.

python

write("claims/db.py", '''
    import sqlite3
    from claims.seed import EXCEPTIONS_MASTER, SHIPMENTS, TMS_CUSTOMERS

    SCHEMA = """
    PRAGMA foreign_keys = ON;
    CREATE TABLE IF NOT EXISTS customers (code TEXT PRIMARY KEY, tms_name TEXT NOT NULL);
    CREATE TABLE IF NOT EXISTS shipments (ref TEXT PRIMARY KEY, customer TEXT NOT NULL REFERENCES customers(code),
        promised TEXT, delivered TEXT, freight_charge REAL);
    CREATE TABLE IF NOT EXISTS exceptions (id INTEGER PRIMARY KEY, ref TEXT NOT NULL REFERENCES shipments(ref),
        type TEXT, owner TEXT, opened TEXT, closed TEXT, message_id TEXT UNIQUE);
    """

    def connect(url):
        db = sqlite3.connect(url.removeprefix("sqlite:///"), check_same_thread=False)
        db.executescript(SCHEMA)
        if db.execute("SELECT COUNT(*) FROM customers").fetchone()[0] == 0:      # seed an empty database once
            db.executemany("INSERT INTO customers VALUES (?, ?)", TMS_CUSTOMERS.items())
            db.executemany("INSERT INTO shipments VALUES (?, ?, ?, ?, ?)", [(ref, s["customer"], s["promised"],
                           s["delivered"], s["freight_charge"]) for ref, s in SHIPMENTS.items()])
            db.executemany("INSERT INTO exceptions (ref, type, owner, opened, closed) VALUES (?, ?, ?, ?, ?)",
                           [(x["ref"], x["type"], x["owner"], x["opened"], x["closed"]) for x in EXCEPTIONS_MASTER])
            db.commit()
        return db
''')
sys.path.insert(0, str(REPO.resolve()))                       # so this notebook can import the package
from claims.db import connect
db = connect("sqlite:///:memory:")
print({t: db.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0] for t in ("customers", "shipments", "exceptions")})

Code walkthrough

IF NOT EXISTS makes the schema safe to run on every start.
connect turns sqlite:///claims.db into a file path. check_same_thread=False lets FastAPI use the connection from its worker threads.
The seed runs only when the database is empty, so a restart never loads the rows twice.
sys.path.insert lets the notebook import claims the way pytest does from inside the repository.

You should see {'customers': 5, 'shipments': 8, 'exceptions': 5}: the shared data, seeded once.

Step 5: The API: read open rows, record an exception once

The same claim posted twice gives one record, EX-006 (green); callers without the clerk role or without a key are refused (red).

The API answers GET /exceptions with the rows Lesson 3's screen draws, and records an exception from an email once, however often it is asked. The settings, database and queue are attached to app at start-up, so tests can attach their own.

python

write("claims/api.py", '''
    from datetime import date
    from fastapi import Depends, FastAPI, Header, HTTPException, Request
    from claims.lateness import days_late
    from claims.seed import DATA_AS_OF

    app = FastAPI(title="Harbourline claims")

    def role(request: Request, authorization: str | None = Header(default=None)):
        found = request.app.state.config["API_KEYS"].get((authorization or "").removeprefix("Bearer "))
        if found is None:
            raise HTTPException(401, "missing or unknown key")
        return found

    OPEN_ROWS = """SELECT e.ref, e.type, e.owner, c.tms_name, s.freight_charge, e.opened
        FROM exceptions e JOIN shipments s ON s.ref = e.ref JOIN customers c ON c.code = s.customer
        WHERE e.closed IS NULL ORDER BY e.opened, e.id"""

    @app.get("/exceptions")                            # the rows Lesson 3's screen draws
    def list_open(request: Request, _: str = Depends(role)):
        rows = request.app.state.db.execute(OPEN_ROWS).fetchall()
        return [{"ref": r, "type": t, "owner": o, "customer": c, "freight": f, "open": True,
                 "age_days": days_late(date.fromisoformat(op), DATA_AS_OF)} for r, t, o, c, f, op in rows]
''')

Code walkthrough

role looks the bearer key up in the settings and raises 401 for an unknown one. Depends(role) runs it before any endpoint that declares it.
OPEN_ROWS is Lesson 2's join; each row takes Lesson 3's Row shape, with age_days counted from DATA_AS_OF.

python

write("claims/intake.py", '''
    import sqlite3
    from typing import Literal
    from fastapi import Depends, HTTPException, Request
    from pydantic import BaseModel
    from claims.api import app, role
    from claims.seed import DATA_AS_OF

    class NewException(BaseModel):
        ref: str
        type: Literal["damage", "late", "missing", "wrong_address"]
        message_id: str

    @app.post("/exceptions", status_code=201)
    def open_exception(body: NewException, request: Request, who: str = Depends(role)):
        if who != "clerk":
            raise HTTPException(403, "clerks only")
        db = request.app.state.db
        try:
            with db:                                   # commit, or roll back on error
                db.execute("INSERT INTO exceptions (ref, type, opened, message_id) VALUES (?, ?, ?, ?)",
                           (body.ref, body.type, DATA_AS_OF.isoformat(), body.message_id))
        except sqlite3.IntegrityError as err:
            if "UNIQUE" not in str(err):               # a foreign key failure: no such shipment
                raise HTTPException(404, f"no shipment {body.ref}")
        row = db.execute("SELECT id, ref, type, message_id FROM exceptions WHERE message_id = ?",
                         (body.message_id,)).fetchone()
        return {"id": f"EX-{row[0]:03d}", "ref": row[1], "type": row[2], "message_id": row[3]}
''')

Code walkthrough

with db: commits the insert, or rolls it back if it fails.
A UNIQUE failure means the email was recorded before, so the stored row is returned. Any other integrity failure is the foreign key: no such shipment, 404.

python

import fakeredis
from fastapi.testclient import TestClient
from claims.config import load_config, safe_to_log
from claims.api import app
import claims.intake                                          # adds POST /exceptions to app

ENV = {"DATABASE_URL": "sqlite:///:memory:", "REDIS_URL": "redis://unused-in-colab",
       "API_KEYS": "key-joan-demo:clerk,key-board-demo:viewer", "MODEL": MODELS[PROVIDER]}   # demo values
app.state.config = load_config(ENV)
app.state.db = connect(ENV["DATABASE_URL"])
app.state.queue = fakeredis.FakeRedis(decode_responses=True)   # stands in for Redis
print(safe_to_log(app.state.config))

client, JOAN = TestClient(app), {"Authorization": "Bearer key-joan-demo"}
claim = {"ref": "HF-88102", "type": "late", "message_id": "MSG-3101"}
for attempt in (1, 2):
    print("attempt", attempt, client.post("/exceptions", json=claim, headers=JOAN).json())
print("viewer:", client.post("/exceptions", json=claim, headers={"Authorization": "Bearer key-board-demo"}).status_code,
      "| no key:", client.get("/exceptions").status_code)
for row in client.get("/exceptions", headers=JOAN).json():
    print(" ", row["ref"], row["type"], row["owner"], row["customer"], row["age_days"])

Code walkthrough

ENV plays the .env file with demo keys. fakeredis stands in for Redis, so REDIS_URL is unused here.
The app.state lines do what the service does at start-up (step 8). The claim is posted twice, as a retry would.

You should see the settings with 'API_KEYS': '****', then both attempts returning EX-006: one record. The viewer gets 403, no key 401. The open rows follow, oldest first: HF-88207 (5 working days open), HF-88190 (1), HF-88301 (0) and the new HF-88102 (0, no owner yet), with TMS customer names; matching them to the CRM is Module 4's job.

Step 6: Triage in the background

The clerk gets job-1 at once (green); the slow model call (amber) happens later in the worker, with the job parked in processing so a crash cannot lose it.

Triage calls the model, which can take seconds, so the API only queues it: POST /messages/{id}/triage answers 202 with a job id, and a worker later asks the model for the type. The model is passed in to the worker, which lets step 7 test it without a key.

python

write("claims/triage.py", '''
    from typing import Literal
    from fastapi import Depends, HTTPException, Request
    from pydantic import BaseModel
    from claims.api import app, role
    from claims.seed import INBOX

    class Triage(BaseModel):
        type: Literal["damage", "late", "missing", "wrong_address"]

    @app.post("/messages/{message_id}/triage", status_code=202)
    def request_triage(message_id: str, request: Request, _: str = Depends(role)):
        q = request.app.state.queue
        job_id = f"job-{q.incr('job_seq')}"
        q.hset(job_id, mapping={"status": "queued", "message_id": message_id})
        q.lpush("triage_jobs", job_id)
        return {"job_id": job_id, "status": "queued"}

    @app.get("/jobs/{job_id}")
    def job_status(job_id: str, request: Request, _: str = Depends(role)):
        return request.app.state.queue.hgetall(job_id) or {"status": "no such job"}

    def work_one(queue, classify):
        job_id = queue.lmove("triage_jobs", "processing", "RIGHT", "LEFT")   # park a copy while working
        if job_id is None:
            return None
        body = next(m["body"] for m in INBOX if m["id"] == queue.hget(job_id, "message_id"))
        queue.hset(job_id, mapping={"status": "done", "type": classify(body)})
        queue.lrem("processing", 1, job_id)
        return job_id
''')

Code walkthrough

Triage limits the model's answer to the four types.
request_triage is Lesson 2's 202 endpoint.
work_one uses lmove, Lesson 2's safer take: the job sits in processing until it is finished, so a crash does not lose it.
classify is a parameter: the worker does not know which model, or whether there is one.

python

import claims.triage                                          # adds the triage routes to app
from claims.triage import Triage, work_one

triager = llm.with_structured_output(Triage, method="json_schema")

def ask(model, messages):
    """Returns (reply, None), or (None, reason) after two refusals."""
    for attempt in range(2):                     # one retry: the next reply is usually fine
        try:
            return model.invoke(messages), None
        except Exception as err:
            last = err
    return None, f"provider_error: {type(last).__name__}"

def classify(text):
    reply, why = ask(triager, "Classify this freight claim email.\n\n" + text)
    return reply.type if reply else why                       # refused twice: the job shows the reason

job = client.post("/messages/MSG-3101/triage", headers=JOAN).json()
print("API answered:", job)
done = work_one(app.state.queue, classify)
print("after the worker:", client.get(f"/jobs/{done}", headers=JOAN).json())

Code walkthrough

with_structured_output(..., method="json_schema") makes the model return a Triage object; this method works on both Groq and Gemini.
classify makes the call a plain function from text to type, the only shape work_one needs; ask retries a refusal once, else returns the reason.

You should see API answered: {'job_id': 'job-1', 'status': 'queued'} straight away, then after the worker: {'status': 'done', 'message_id': 'MSG-3101', 'type': 'late'}. Both Groq and Gemini returned late when this lab was tested.

Step 7: Every test, run as CI runs them

Passing the model in means CI tests the same worker with a stand-in (green): eight tests pass with no key and no secrets to steal.

The API tests build their own app state: an in-memory database, fakeredis and a stand-in classifier, lambda text: "late". That means CI needs no model key and no secrets at all, which is the safest number of secrets a pipeline can hold.

python

write("tests/test_api.py", '''
    import fakeredis
    from fastapi.testclient import TestClient
    from claims.config import load_config
    from claims.db import connect
    from claims.api import app
    import claims.intake, claims.triage
    from claims.triage import work_one

    ENV = {"DATABASE_URL": "sqlite:///:memory:", "REDIS_URL": "redis://unused", "API_KEYS": "k-clerk:clerk", "MODEL": "none"}
    app.state.config, app.state.db = load_config(ENV), connect(ENV["DATABASE_URL"])
    app.state.queue = fakeredis.FakeRedis(decode_responses=True)
    client, CLERK = TestClient(app), {"Authorization": "Bearer k-clerk"}

    def test_open_rows_need_a_key():
        assert client.get("/exceptions").status_code == 401
        assert [r["ref"] for r in client.get("/exceptions", headers=CLERK).json()] == ["HF-88207", "HF-88190", "HF-88301"]

    def test_a_retry_records_once():
        claim = {"ref": "HF-88102", "type": "late", "message_id": "MSG-3101"}
        assert len({client.post("/exceptions", json=claim, headers=CLERK).json()["id"] for _ in range(2)}) == 1

    def test_triage_runs_without_a_model():
        job = client.post("/messages/MSG-3101/triage", headers=CLERK).json()["job_id"]
        assert work_one(app.state.queue, lambda text: "late") == job
        assert client.get(f"/jobs/{job}", headers=CLERK).json()["type"] == "late"
''')
print(ci())

Code walkthrough

test_open_rows_need_a_key checks the 401 and the three open rows, oldest first.
test_a_retry_records_once collects the ids of two posts in a set; one id means one record.
test_triage_runs_without_a_model queues MSG-3101, runs the worker with the stand-in and reads the job back.

You should see exit code 0: 8 passed. Eight tests, no key, no network.

Step 8: Package it for Harbourline's VM

The image is built from the repository; Compose runs it beside Redis on Harbourline's VM, with the data in a volume and .env read on the VM, never shipped (grey is supporting detail).

The last step writes the files that take the slice off your laptop: the start-up module, requirements, Dockerfile, Compose file and CI workflow, adjusted from Lesson 4. Then the first commit. Docker does not run in Colab, so the containers start on your own computer.

python

write("claims/main.py", '''
    import redis
    from claims.api import app
    from claims.config import load_config, safe_to_log
    from claims.db import connect
    import claims.intake, claims.triage                # registers every route

    app.state.config = load_config()                   # refuses to start without its settings
    print("starting with", safe_to_log(app.state.config), flush=True)
    app.state.db = connect(app.state.config["DATABASE_URL"])
    app.state.queue = redis.Redis.from_url(app.state.config["REDIS_URL"], decode_responses=True)
''')
write("requirements.txt", "fastapi[standard]\nredis\nlangchain\nlangchain-groq\nlangchain-google-genai\n")
write("Dockerfile", '''
    FROM python:3.14-slim
    WORKDIR /app
    COPY requirements.txt .
    RUN pip install --no-cache-dir -r requirements.txt
    COPY claims/ claims/
    RUN useradd --create-home appuser && mkdir /data && chown appuser /data
    USER appuser
    EXPOSE 8000
    CMD ["fastapi", "run", "claims/main.py", "--port", "8000"]
''')

Code walkthrough

main.py does at start-up what the notebook did by hand: load settings (or stop), log them safely, connect the database and the real Redis. flush=True makes the log line appear at once inside a container.
The Dockerfile is Lesson 4's: a slim Python base, requirements before code, a non-root user. /data is a folder the app user owns, for the SQLite file.

python

write("compose.yaml", '''
    services:
      api:
        build: .
        ports: ["127.0.0.1:8000:8000"]
        env_file: .env
        environment:
          DATABASE_URL: sqlite:////data/claims.db
          REDIS_URL: redis://redis:6379/0
        volumes: ["claims_data:/data"]
        depends_on: [redis]
      redis:
        image: redis:7
    volumes:
      claims_data:
''')
write(".github/workflows/ci.yml", '''
    name: ci
    on:
      push:
        branches: [main]
      pull_request:
    jobs:
      test:
        runs-on: ubuntu-latest
        steps:
          - uses: actions/checkout@v7
          - uses: actions/setup-python@v7
            with: {python-version: "3.14"}
          - run: pip install -r requirements.txt pytest fakeredis
          - run: pytest -q
          - run: docker build -t claims-api:${{ github.sha }} .
''')
write(".env", "API_KEYS=key-joan-local:clerk\nMODEL=groq:openai/gpt-oss-20b\nGROQ_API_KEY=paste-your-own\n")
git("add", ".")
print(git("commit", "-m", "Claims service skeleton: policy, settings, store, API, triage, CI").splitlines()[0])
print(".env on disk:", (REPO / ".env").exists(), "| tracked by Git:", ".env" in git("ls-files").splitlines())

Code walkthrough

compose.yaml runs the API beside Redis. The API reaches Redis by its service name, redis, never localhost (Lesson 4's break). The port is published on the VM's own address only, and the volume keeps the database when the container is replaced.
ci.yml is Lesson 4's pipeline. It holds no secrets. line, because the tests need none.
.env gets local values. git add . stages everything; the last line checks that .env stayed out.

You should see the first commit line, [main (root-commit) ...] Claims service skeleton: policy, settings, store, API, triage, CI (the commit id differs every run), then .env on disk: True | tracked by Git: False.

On your own computer, with Docker installed and the repository copied down from Colab (not run here, verified against the Docker Compose CLI reference, https://docs.docker.com/reference/cli/docker/compose/):

bash

cd harbourline-claims
docker compose up -d --build
curl -s -H "Authorization: Bearer key-joan-local" http://127.0.0.1:8000/exceptions
docker compose logs api | grep "starting with"
docker compose down

Code walkthrough

up -d --build builds the image and starts both containers. curl sends one request with your .env key and should return the three seeded open rows.
The grep finds the start-up line, where API_KEYS shows as ****. down stops the containers; the volume stays.

Check your work

Check	Guards against	Result
tests pass	a broken rule reaching main	PASS
refuses to start	a half-configured VM	PASS
.env not tracked	a key in the repository	PASS
no key in history	a key in an old commit	PASS
files present	nothing to deploy	PASS

Five checks, each guarding one failure this module taught; all five pass on the finished slice.

Five checks cover what this module promised. The start-up check runs main.py in a separate Python with the settings removed, as on a half-configured VM.

python

def refuses_to_start_without_settings():
    env = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "REDIS_URL", "API_KEYS", "MODEL")}
    r = subprocess.run([sys.executable, "-B", "-c", "import claims.main"], cwd=REPO, env=env,
                       capture_output=True, text=True)
    return r.returncode != 0 and "missing settings: DATABASE_URL, REDIS_URL, API_KEYS, MODEL" in r.stderr

checks = {
    "tests pass": ci().startswith("exit code 0"),
    "refuses to start without settings": refuses_to_start_without_settings(),
    ".env exists and is not tracked": (REPO / ".env").exists() and ".env" not in git("ls-files").splitlines(),
    "no local key anywhere in Git history": "key-joan-local" not in git("log", "-p", "--all"),
    "Dockerfile, compose.yaml and CI present": all((REPO / p).exists() for p in ("Dockerfile", "compose.yaml", ".github/workflows/ci.yml")),
}
for name, passed in checks.items():
    print("PASS" if passed else "FAIL", name)

Code walkthrough

refuses_to_start_without_settings passes only if importing claims.main fails with an error naming all four settings.
git log -p --all prints every change ever committed; the local key must not be in it.

You should see five lines, each starting PASS.

Break it

Removing .env in a later commit leaves the key in history (red), where git log -S finds it; the latest commit only looks clean (amber), and the real fix is rotation (green).

Someone tidies .gitignore and drops the .env line, commits, notices, and removes .env in the next commit. The file is gone from the latest commit. Run the history check again.

python

write(".gitignore", "__pycache__/\n*.pyc\n*.db\n")                  # the .env line is gone
git("add", "."); git("commit", "-m", "Tidy ignore file")
git("rm", "--cached", "--quiet", ".env"); git("commit", "-m", "Remove .env")
print("tracked now:", ".env" in git("ls-files").splitlines())
print("key in history:", "key-joan-local" in git("log", "-p", "--all"))
print(git("log", "--oneline", "-S", "key-joan-local"))
git("reset", "--hard", "HEAD~2")                                   # local only: never after a push
print("after reset:", "key-joan-local" in git("log", "-p", "--all"))

Code walkthrough

git rm --cached stops tracking .env but leaves the file on disk.
git log -S lists every commit that added or removed the text.
git reset --hard HEAD~2 moves main back two commits, so --all no longer sees them. That is only honest before anyone has pulled.

You should see tracked now: False, yet key in history: True. The -S search lists both commits, Remove .env and Tidy ignore file. After the reset, after reset: False.

Deleting a secret from the latest commit does not delete it. Anyone with a copy of the repository has every earlier commit. Once a key has been committed and pushed, the fix is to rotate it at the source, the CRM or the identity provider, and treat the old value as public. The incident file shows what that costs when it is a whole platform's secrets.

What you added to the engagement

harbourline-claims: one repository with a policy module, settings that fail fast, a seeded SQLite store, an API with roles, idempotent intake and background triage.
Eight tests that run in CI with no model key and no secrets.
A Dockerfile, a compose.yaml and a CI workflow, ready for Viktor's VM.
A .gitignore and .env.example that keep credentials out of Git, and a check that proves it.

Going further

Point Lesson 3's React screen at this API through its Vite proxy; its fetch now needs a key, which Module 6 replaces with single sign-on.
Add the PostgreSQL service from Lesson 4 to compose.yaml and a connect that reads postgresql:// URLs.
Run work_one in a loop as a second Compose service, worker, using MODEL from the settings.
Add a CI step that scans the repository history for secrets before the tests run.