Score hospital readmissions asynchronously

What you will be able to do

Serve a model whose work does not fit in one request. A hospital’s readmission model scores a large discharge file as a queued job. The caller gets an answer at once, asks again later, and fetches the follow-up list when it is ready. Then you make the jobs survive a restart, because a container restarts more often than you think.

Case studyPortfolio piece

Dataset

Diabetes 130-US hospitals, 1999–2008 (UCI)

Licence

CC BY 4.0 (data); MIT (code)

Time

About 3 hours

Runs in

VS Code, Python 3.12+

Skills used

• Asynchronous jobs
• 202 Accepted and polling
• Queue and size limits
• Partial failure
• Result retention
• Durable job storage
• Containers and volumes

Download the starter project (.zip)Finished project (.zip)

Unzip the starter, open the folder in VS Code, and follow the steps below. They start with a new virtual environment.

The starter is a small service with a checked model bundle and a first, synchronous endpoint. You replace that endpoint with jobs, put the service in a container, break it with a restart, and fix it.

Before you startWhat to know and to have readyAbout 180 minutes

What you need to know

Asynchronous inference has the lead role. The other lessons that this case study uses most:

• The bundle, its digest and its six gates, from Artifact trust and integrity.
• Start-up work and readiness, from Load and initialize.
• The Dockerfile, Compose and resource limits, from Container resource settings.
• Volumes, from the Docker course: Persistent data.

You do not need the machine learning course. If you did its case study Choose a model for hospital readmissions, this is the model that it chose: now you serve it.

What you need on your computer

• Python 3.12 or newer, VS Code with the Python extension, and curl.
• Docker Desktop, as in Container resource settings.
• About 1.2 GB of free disk space: 360 MB for the virtual environment and 790 MB for the image.

Tested with Python 3.14.6 and Docker Desktop (Docker Engine 29.6.1) on macOS, Apple silicon. The packages: FastAPI 0.142.2, scikit-learn 1.9.1, pandas 3.0.6, NumPy 2.5.3, joblib 1.6.0 and pytest 9.1.1. The Windows and Linux steps are not tested.

The situation

Riverbend Health runs a network of hospitals. Its care-coordination team, led by Adaeze Obi, phones patients in the first days after they leave hospital. A nurse checks the medicines, books a follow-up visit and asks who helps at home. The team can call about one discharged patient in ten, so it needs a list: who first?

A model already scores each hospital stay for the risk of a readmission within 30 days (the case study of the machine learning course). Rafael Costa, a data engineer, exports the discharges each night at 02:00 as one CSV file. The nurses start at 07:30. Jun Park, a developer, builds the service between the two. You are Jun.

Imani Brooks, the clinical safety officer, sets three conditions. The list only says whom to offer a call first: it never decides anything about a patient on its own. The service receives only the columns that the model needs. And no list is kept longer than 48 hours.

Rafael asks: “My export can send the file and wait for the answer. Why would we need anything else?” Part 2 answers with measurements.

What you will build

A service with three job endpoints, with the course’s conventions (a checked bundle, one error shape, an API key, /health and /ready, a container with limits).

Endpoint	What it does
POST /v1/jobs	Accept a discharge file (CSV) as a job: 202 Accepted at once, with the job’s address
GET /v1/jobs/{id}	The job’s status: queued, running (with progress), done, failed or expired
GET /v1/jobs/{id}/results	The follow-up list (CSV), the highest score first, for 48 hours
GET /v1/model	The model version, its bundle digest and its threshold

Each part starts from a problem: a slow file, a burst of files, a broken export, a list kept too long, a restart.

The model and its data

The data is the Diabetes 130-US Hospitals for Years 1999–2008 dataset (Clore, Cios, DeShazo and Strack, 2014; UCI Machine Learning Repository, doi:10.24432/C5230J), under the CC BY 4.0 licence. It holds 101,766 hospital stays of patients with diabetes in 130 US hospitals. It is de-identified, but it is still health data about real people: never try to find out who anyone is. Riverbend Health and its team are invented.

The model is the one that the machine learning case study chose: a logistic regression on 19 features that are known on the day of discharge (for example how the patient came in, where they go next, the main diagnosis group and the medicines). Race, gender and the payer are not inputs. It learned from 59,556 stays of 41,999 patients, split by patient.

The threshold is 0.18: the score at the 10% mark of the validation stays, for the team’s capacity of one call in ten.

Set	ROC AUC	Called (score ≥ 0.18)	Precision of the calls	Recall
Valid, 20,179 stays	0.671	2,013 (10.0%)	0.261	0.231
Test, 19,608 stays (used once)	0.663	1,866 (9.5%)	0.270	0.221

About 27 calls in 100 reach a patient who comes back (at random, about 11). But they reach only 22% of those patients: the list helps the nurses, it does not replace them. (The machine learning case study reports 1,870 test calls with the unrounded threshold.)

The discharge files are the 19,608 test stays: a week (all of them, 2.8 MB) and a night (the first seventh, 2,801 stays). The data has no dates, so “week” and “night” are a teaching device. A file has the stay’s encounter_id and the 19 features: no names, no patient numbers.

Set up the starter project

1. Download the starter project with the button in the box at the top of this page. Unzip it into your projects folder. You get a folder readmission-service.

2. In VS Code, open the folder: File › Open Folder…, then choose readmission-service.

3. Open a terminal in VS Code: Terminal › New Terminal. It starts in the project folder.

4. Make and activate a virtual environment:

   macOS

   python3 -m venv .venv
   source .venv/bin/activate

   Windows

   python -m venv .venv
   .venv\Scripts\Activate.ps1

   Linux

   python3 -m venv .venv
   source .venv/bin/activate

   The prompt starts with (.venv).

5. Install the exact versions of the packages:

   python -m pip install -r requirements.txt

   This takes a minute or two. The environment uses about 360 MB.

6. Tell VS Code to use this environment. Open the Command Palette (Ctrl+Shift+P, or Cmd+Shift+P on macOS), choose Python: Select Interpreter, then choose the one in .venv.

7. Run the tests:

   python -m pytest

The output (last lines):

collected 19 items

tests/test_bundle.py ......                                              [ 31%]
tests/test_contract.py .........                                         [ 78%]
tests/test_health.py ....                                                [100%]

============================= 19 passed in 15.43s ==============================

The first run is slow because Python compiles the packages; the next runs take about 2 seconds.

Path	What it is
bundles/readmission-1.0.0/	The model bundle: model.joblib (12 KB), contract.json, metadata.json, parity_cases.csv (10 cases), SHA256SUMS
readmission/	contract.py (the 19 features and their rules), bundle.py (the six gates), scoring.py (check and score a file), settings.py, errors.py, security.py, main.py, routes.py
data/	The discharge files, two broken files, and outcomes_week.csv (what happened, for checks only); data/README.md says how they were made
training/train.py, make_bundle.py	How the model and the bundle were made
scripts/follow_up.py	The care team’s client: submit a file, wait, save the list (you use it in Part 4)
MODEL_CARD.md, LICENSE	What the model is for and not for; the MIT licence of the code

Part 1: check the bundle before you serve it

Loading a joblib file can run code. So the service first checks the bundle’s digest against a value from outside the bundle, as in Artifact trust and integrity. Compute the digest and compare it with the one in .env.example:

macOS

shasum -a 256 bundles/readmission-1.0.0/SHA256SUMS
grep MODEL_SHA256 .env.example

Windows

(Get-FileHash bundles\readmission-1.0.0\SHA256SUMS).Hash.ToLower()
Select-String MODEL_SHA256 .env.example

Linux

sha256sum bundles/readmission-1.0.0/SHA256SUMS
grep MODEL_SHA256 .env.example

The output (macOS):

89056e4cfab4a6a58180461ab01f8b53aeec448ed444ee808699abe2ed6cdf27  bundles/readmission-1.0.0/SHA256SUMS
MODEL_SHA256=89056e4cfab4a6a58180461ab01f8b53aeec448ed444ee808699abe2ed6cdf27

In a real team, the digest comes from the model’s release record, not from a file next to the bundle. Here, this page is that record.

For provenance, trace the bundle back to its source: training/train.py downloads the data (3 MB), checks its SHA-256, and trains the model again. Run it once:

python training/train.py

The output:

Downloading diabetes_130_hospitals.csv.gz...
Downloading IDS_mapping.csv...
Trained on 59556 stays of 41999 patients. Threshold 0.18 for a call capacity of 10%.
Valid: ROC AUC 0.671, average precision 0.206, called 2013 of 20179, precision 0.261, recall 0.231.
Test: ROC AUC 0.663, average precision 0.217, called 1866 of 19608, precision 0.27, recall 0.221.
Model file SHA-256: a7fbea052900df4b3750342a3400870914f756c4a25a91aee571150d4284baee

The last line equals the line for model.joblib in SHA256SUMS: the same data and code give the same file, byte for byte. Anyone can check that the bundle holds the model that the machine learning case study describes. (The script also writes the data files again, with the same contents.)

Part 2: does this file need a job?

The dry cleaner’s ticket of Asynchronous inference is worth its cost only when the wait is long. So first, measure the wait.

The honest number

The starter has Rafael’s idea: POST /v1/follow-up-list scores the file while the caller waits. Start the service in the first terminal:

macOS

export MODEL_SHA256=89056e4cfab4a6a58180461ab01f8b53aeec448ed444ee808699abe2ed6cdf27
fastapi dev

Windows

$env:MODEL_SHA256 = "89056e4cfab4a6a58180461ab01f8b53aeec448ed444ee808699abe2ed6cdf27"
fastapi dev

Linux

export MODEL_SHA256=89056e4cfab4a6a58180461ab01f8b53aeec448ed444ee808699abe2ed6cdf27
fastapi dev

Open a second terminal (Terminal › New Terminal) and send the night file, then the week file. On Windows, type curl.exe instead of curl.

Second terminal

mkdir out
curl -s -o out/night.csv -w "%{http_code} in %{time_total} s\n" -X POST --data-binary @data/discharges_night.csv -H "Content-Type: text/csv" http://127.0.0.1:8000/v1/follow-up-list
curl -s -o out/week.csv -w "%{http_code} in %{time_total} s\n" -X POST --data-binary @data/discharges_week.csv -H "Content-Type: text/csv" http://127.0.0.1:8000/v1/follow-up-list
python -c "print(''.join(open('out/week.csv').readlines()[:4]), end='')"

The output:

200 in 0.055770 s
200 in 0.218660 s
rank,encounter_id,score,call
1,291808098,0.876674,true
2,33695262,0.818332,true
3,26662614,0.807009,true

The real model scores the whole week in about 0.2 seconds (0.19 to 0.22 s in our runs). With this model, Rafael is right: if nothing changes, keep the simple synchronous endpoint. A job adds a store, polling, retention and more ways to fail.

What changes the answer

Three facts at Riverbend change it:

• The team plans a slower model that also reads each discharge letter. We do not have it, so we simulate its cost: EXTRA_MS_PER_RECORD adds that many milliseconds per row (0 in production).
• A gateway closes any request after 30 seconds. curl --max-time 30 does the same.
• Nobody waits at 02:00. If the service restarts during the night, the list must still be there at 07:30.

Stop the service (Ctrl+C) and start it with 5 ms more per row:

macOS

EXTRA_MS_PER_RECORD=5 fastapi dev

Windows

$env:EXTRA_MS_PER_RECORD = "5"; fastapi dev

Linux

EXTRA_MS_PER_RECORD=5 fastapi dev

The log says WARNING readmission: SIMULATED slow model: 5.0 ms more per row. Send the week again, with the gateway’s limit:

Second terminal

curl -sS -o out/week.csv --max-time 30 -X POST --data-binary @data/discharges_week.csv -H "Content-Type: text/csv" http://127.0.0.1:8000/v1/follow-up-list

The output, after 30 seconds:

curl: (28) Operation timed out after 30005 milliseconds with 0 bytes received

About a minute later, the first terminal shows:

INFO readmission: POST /v1/follow-up-list 200 99027.1ms id=3d3ad2e97791

The client gave up after 30 seconds, but the service did not know: it answered 200 after 99 seconds, to nobody. The caller has no list, the service spent the work, and a retry would do it all again.

Where it stops fitting

Predict: with how many milliseconds more per row does the week file stop fitting in 30 seconds?

The time is about the rows times the time per row: 19,608 × 1.5 ms = 29.4 s. For the night, even 5 ms per row gives 2,801 × 5 ms = 14 s. The chart shows measured job times:

• Week file
• Night file
• 30 s limit

Measured job times for the night file (2,801 rows) and the week file (19,608 rows). At 0, the real model: 0.04 s and 0.36 s. Above 0, a simulated slower model: each millisecond per row adds about 20 s to the week.

Text description of the chart

A line chart. The horizontal axis is the extra time per row in milliseconds, from 0 to 5. 0 is the real model; 1, 2 and 5 are simulated. The vertical axis is the measured job time in seconds, from 0 to 100. The week file (19,608 rows) takes 0.36 s with the real model. With 1, 2 and 5 ms more per row, it takes 20.1 s, 40.2 s and 98.8 s. The night file (2,801 rows) takes 0.04 s, 2.9 s, 5.8 s and 14.1 s. A dashed horizontal line marks the 30-second gateway limit. The week line crosses it between 1 and 2 ms per row; the night line stays below it everywhere on the chart.

The choice depends on the product of the file size and the time per row, not on either alone. Jun builds the job for the files and models that are coming, and keeps the measurement to justify it. The cost, as in module 2: a store, polling, retention, more states, and about 270 lines of code (120 more to survive a restart).

Part 3: accept the file as a job

The life of a job

A job turns one long request into several short ones, as in module 2. Here the worker is a thread inside the service, and the caller polls for the status.

The life of a job. The request does only the fast checks; the worker does the scoring; the caller asks again, never more often than Retry-After allows.

Text description of the diagram

Read it from top to bottom. The two answers that matter are 202 Accepted, which promises nothing except that the job exists, and Retry-After, which tells a polite client when to ask again. Every slow step is on the worker’s line, so no request waits for the model.

Four rules come with it, each a line of code below:

• Check what is fast before you accept: a missing column gets an error now, not a failed job later.
• Limit the queue: when 3 jobs wait, the next one gets 429 Too Many Requests.
• A failed job must not stop the worker.
• Keep the list for a stated time: after 48 hours, the job says expired.

Step 1: the jobs module

Make a new file, readmission/jobs.py. It has three parts: the Job record, a MemoryStore that keeps jobs in the process’s memory, and the JobRunner with its worker thread. The first part:

readmission/jobs.py

"""Asynchronous scoring: accept a discharge file now, score it later, keep the list for a while.

    submit     check the queue limit, store the file, return the job at once
    work       one thread takes the oldest queued job and runs it
    run        check the rows, score them in chunks, store the follow-up list

A job is queued → running → done or failed, and expired when its time is up.
The runner keeps nothing itself: every job, file and list is in the store.
MemoryStore keeps them in this process's memory, so a restart loses them all.
"""

import hashlib
import logging
import threading
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone

from readmission.scoring import read_rows, results_csv, score_rows

logger = logging.getLogger("readmission")

CHUNK = 1000  # rows scored between two progress updates
MAX_REJECTED_SHARE = 0.05  # more bad rows than this: the file is wrong, not a few rows


class QueueFull(Exception):
    pass


class JobFailed(Exception):
    pass


def now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


@dataclass
class Job:
    id: str
    records: int
    input_sha256: str
    status: str = "queued"
    created_at: str = field(default_factory=now)
    started_at: str | None = None
    finished_at: str | None = None
    expires_at: str | None = None
    scored: int = 0
    rejected: list[dict] = field(default_factory=list)
    to_call: int | None = None
    unknown_values: dict[str, int] = field(default_factory=dict)
    seconds: float | None = None
    model_version: str | None = None
    error: str | None = None


class MemoryStore:
    """Jobs, files and lists in this process's memory. A restart loses every job."""

    def __init__(self):
        self.jobs: dict[str, Job] = {}
        self.files: dict[str, bytes] = {}
        self.lists: dict[str, str] = {}
        self.lock = threading.Lock()

    def add(self, job: Job, data: bytes) -> None:
        with self.lock:
            self.jobs[job.id], self.files[job.id] = job, data

    def get(self, job_id: str) -> Job | None:
        return self.jobs.get(job_id)

    def save(self, job: Job) -> None:
        self.jobs[job.id] = job

    def queued(self) -> int:
        return sum(job.status == "queued" for job in self.jobs.values())

    def claim_next(self) -> Job | None:
        with self.lock:
            waiting = [job for job in self.jobs.values() if job.status == "queued"]
            if not waiting:
                return None
            job = min(waiting, key=lambda j: j.created_at)
            job.status, job.started_at = "running", now()
            return job

    def file(self, job_id: str) -> bytes:
        return self.files[job_id]

    def finish(self, job: Job, follow_up_list: str | None) -> None:
        """Keep the list (if any), and drop the input file: it is not needed any more."""
        with self.lock:
            if follow_up_list is not None:
                self.lists[job.id] = follow_up_list
            self.files.pop(job.id, None)
            self.jobs[job.id] = job

    def follow_up_list(self, job_id: str) -> str | None:
        return self.lists.get(job_id)

    def expire(self) -> int:
        """Delete the lists whose time is up. The job stays, as "expired"."""
        count = 0
        with self.lock:
            for job in self.jobs.values():
                if job.expires_at and job.expires_at <= now() and job.status != "expired":
                    job.status = "expired"
                    self.lists.pop(job.id, None)
                    count += 1
        return count

    def recover(self) -> int:
        return 0  # nothing survives a restart, so there is nothing to recover

The store is a separate class on purpose. In Part 7 you replace it with one that survives a restart, and the runner does not change. Add the runner at the end of the file:

readmission/jobs.py (continued)

class JobRunner:
    def __init__(self, store, get_bundle, max_queued: int, keep_hours: float, extra_ms: float = 0.0):
        self.store, self.get_bundle = store, get_bundle
        self.max_queued, self.keep, self.extra_ms = max_queued, timedelta(hours=keep_hours), extra_ms
        self.wake = threading.Event()
        threading.Thread(target=self.work, daemon=True, name="job-worker").start()

    def submit(self, data: bytes, records: int) -> Job:
        if self.store.queued() >= self.max_queued:
            raise QueueFull
        job = Job(id=uuid.uuid4().hex[:12], records=records, input_sha256=hashlib.sha256(data).hexdigest())
        self.store.add(job, data)
        self.wake.set()
        return job

    def work(self) -> None:
        while True:
            self.store.expire()
            job = self.store.claim_next()
            if job is None:
                self.wake.wait(timeout=60)
                self.wake.clear()
                continue
            self.run(job)

    def run(self, job: Job) -> None:
        started, follow_up = time.perf_counter(), None
        try:
            rows, job.rejected = read_rows(self.store.file(job.id))
            if len(job.rejected) > MAX_REJECTED_SHARE * job.records:
                raise JobFailed(f"{len(job.rejected)} of {job.records} rows do not match the contract")
            bundle, results = self.get_bundle(), []
            for start in range(0, len(rows), CHUNK):
                results += score_rows(bundle, rows[start : start + CHUNK], self.extra_ms)
                job.scored = len(results)
                self.store.save(job)  # progress, for the status
            follow_up = results_csv(results)
            job.to_call = sum(r.call for r in results)
            for r in results:
                for name in r.unknown:
                    job.unknown_values[name] = job.unknown_values.get(name, 0) + 1
            job.model_version, job.status = bundle.version, "done"
        except JobFailed as error:
            job.status, job.error = "failed", str(error)
        except Exception:  # a failed job must not stop the worker
            logger.exception("job %s failed", job.id)
            job.status, job.error = "failed", "The job failed on the server. The log has the details."
        job.seconds = round(time.perf_counter() - started, 3)
        job.finished_at = now()
        job.expires_at = (datetime.now(timezone.utc) + self.keep).isoformat(timespec="seconds")
        self.store.finish(job, follow_up)
        logger.info("job %s %s: %d rows, %.2f s", job.id, job.status, job.records, job.seconds)

submit raises QueueFull when max_queued jobs wait. run saves scored after each chunk of 1,000 rows, so the status shows progress. And except Exception turns any error into a failed job, so the worker goes on to the next one.

Step 2: the routes

Open readmission/routes.py. Replace the imports at the top with these:

readmission/routes.py

from fastapi import APIRouter, Depends, Request, Response
from pydantic import BaseModel, Field

from readmission.errors import ErrorResponse, error_response
from readmission.jobs import Job, QueueFull, now
from readmission.scoring import FileProblem, check_upload
from readmission.security import require_api_key

Then delete the whole follow_up_list function, from @router.post( to its last line, and add the job routes at the end of the file:

readmission/routes.py (at the end)

class JobStatus(BaseModel):
    job_id: str
    status: str = Field(description="queued, running, done, failed or expired")
    records: int = Field(description="Rows in the file.")
    scored: int = Field(description="Rows scored so far.")
    rejected: int = Field(description="Rows that do not match the contract. They are not scored.")
    rejected_lines: list[dict] = Field(description="The first 20 rejected rows: line in the file, and fields.")
    to_call: int | None = Field(description="Stays with a score at or above the threshold.")
    unknown_values: dict[str, int] = Field(description="Category values the model never saw, by column.")
    created_at: str
    finished_at: str | None
    expires_at: str | None = Field(description="After this time, the list is deleted.")
    seconds: float | None
    model_version: str | None
    results_url: str | None
    error: str | None


def is_expired(job: Job) -> bool:
    return job.status == "expired" or (job.expires_at is not None and job.expires_at <= now())


def status_of(job: Job) -> JobStatus:
    expired = is_expired(job)
    return JobStatus(
        job_id=job.id,
        status="expired" if expired else job.status,
        records=job.records,
        scored=job.scored,
        rejected=len(job.rejected),
        rejected_lines=job.rejected[:20],
        to_call=job.to_call,
        unknown_values=job.unknown_values,
        created_at=job.created_at,
        finished_at=job.finished_at,
        expires_at=job.expires_at,
        seconds=job.seconds,
        model_version=job.model_version,
        results_url=f"/v1/jobs/{job.id}/results" if job.status == "done" and not expired else None,
        error=job.error,
    )


@router.post(
    "/jobs",
    tags=["jobs"],
    status_code=202,
    summary="Submit a discharge file (CSV) to score later",
    openapi_extra=CSV_BODY,
    responses=errors(401, 413, 415, 422, 429),
)
async def submit_job(request: Request, response: Response) -> JobStatus:
    settings = request.app.state.settings
    if request.headers.get("content-type", "").split(";")[0].strip() != "text/csv":
        return error_response(request, 415, "not_csv", "Send the file as text/csv.")
    body = await request.body()
    if len(body) > settings.max_upload_bytes:
        return error_response(request, 413, "file_too_large", f"A file can have at most {settings.max_upload_bytes} bytes.")
    try:
        records = check_upload(body, settings.max_records)
        job = request.app.state.jobs.submit(body, records)
    except FileProblem as problem:
        return error_response(request, problem.status, problem.code, problem.message, problem.fields)
    except QueueFull:
        return error_response(
            request, 429, "too_many_jobs", "Too many jobs are waiting. Try again later.", headers={"Retry-After": "60"}
        )
    response.headers["Location"] = f"/v1/jobs/{job.id}"
    response.headers["Retry-After"] = str(settings.poll_seconds)
    return status_of(job)


@router.get("/jobs/{job_id}", tags=["jobs"], summary="The status of a job", responses=errors(401, 404))
def get_job(job_id: str, request: Request, response: Response) -> JobStatus:
    job = request.app.state.jobs.store.get(job_id)
    if job is None:
        return error_response(request, 404, "job_not_found", "There is no job with this ID.")
    if job.status in ("queued", "running"):
        response.headers["Retry-After"] = str(request.app.state.settings.poll_seconds)
    return status_of(job)


@router.get(
    "/jobs/{job_id}/results",
    tags=["jobs"],
    summary="The follow-up list of a finished job (CSV, highest score first)",
    response_class=Response,
    responses={200: {"content": {"text/csv": {}}}, **errors(401, 404, 409, 410)},
)
def get_results(job_id: str, request: Request):
    store = request.app.state.jobs.store
    job = store.get(job_id)
    if job is None:
        return error_response(request, 404, "job_not_found", "There is no job with this ID.")
    if is_expired(job):
        return error_response(request, 410, "job_expired", f"The list expired at {job.expires_at}.")
    if job.status == "failed":
        return error_response(request, 409, "job_failed", f"The job failed: {job.error}")
    if job.status != "done":
        seconds = str(request.app.state.settings.poll_seconds)
        return error_response(request, 409, "job_not_done", "The job is not done yet.", headers={"Retry-After": seconds})
    return Response(store.follow_up_list(job_id), media_type="text/csv")

The checks in submit_job run from the cheapest to the most expensive: header, size, columns and row count, queue. Nothing reads a row’s values: that is the worker’s job.

Step 3: start the runner

In readmission/main.py, add the import below the other readmission imports:

readmission/main.py

from readmission.jobs import JobRunner, MemoryStore

Then, in lifespan, after the logger.info(...) call that reports the loaded model, start the runner:

readmission/main.py

        app.state.jobs = JobRunner(
            MemoryStore(),
            lambda: app.state.bundle,
            settings.max_queued_jobs,
            settings.keep_results_hours,
            settings.extra_ms_per_record,
        )

The runner starts after the bundle passed its gates, so a job can never meet a model that is not ready.

Step 4: try it

Stop the service. Start it again without the simulated cost (on Windows, first Remove-Item Env:EXTRA_MS_PER_RECORD), so fastapi dev. Then submit the sample file:

Second terminal

curl -si -X POST --data-binary @data/discharges_sample.csv -H "Content-Type: text/csv" http://127.0.0.1:8000/v1/jobs

The output:

HTTP/1.1 202 Accepted
date: Thu, 08 Oct 2026 14:54:49 GMT
server: uvicorn
content-length: 280
content-type: application/json
location: /v1/jobs/4a2e2c5b5daa
retry-after: 2
x-request-id: 2c1d01493006

{"job_id":"4a2e2c5b5daa","status":"queued","records":5,"scored":0,"rejected":0,"rejected_lines":[],"to_call":null,"unknown_values":{},"created_at":"2026-10-08T14:54:49+00:00","finished_at":null,"expires_at":null,"seconds":null,"model_version":null,"results_url":null,"error":null}

Ask for the job at its location, with your own job ID, and then for its list:

Second terminal

curl -s http://127.0.0.1:8000/v1/jobs/4a2e2c5b5daa
curl -s http://127.0.0.1:8000/v1/jobs/4a2e2c5b5daa/results

The output:

{"job_id":"4a2e2c5b5daa","status":"done","records":5,"scored":5,"rejected":0,"rejected_lines":[],"to_call":0,"unknown_values":{},"created_at":"2026-10-08T14:54:49+00:00","finished_at":"2026-10-08T14:54:50+00:00","expires_at":"2026-10-10T14:54:50+00:00","seconds":0.022,"model_version":"1.0.0","results_url":"/v1/jobs/4a2e2c5b5daa/results","error":null}
rank,encounter_id,score,call
1,40926,0.144761,false
2,15738,0.110223,false
3,28236,0.095788,false
4,35754,0.080952,false
5,16680,0.071498,false

The 202 came before any scoring. A second later, the job is done, names its model, and says when its list expires. None of the 5 stays reaches 0.18.

Now fill the queue. Restart the service with the simulated 5 ms per row, as in Part 2, so that each night file takes 14 seconds. Then submit the night file five times, quickly, and once more with the headers:

Second terminal

for i in 1 2 3 4 5; do curl -s -o /dev/null -w "%{http_code} " -X POST --data-binary @data/discharges_night.csv -H "Content-Type: text/csv" http://127.0.0.1:8000/v1/jobs; done; echo
curl -si -X POST --data-binary @data/discharges_night.csv -H "Content-Type: text/csv" http://127.0.0.1:8000/v1/jobs

On Windows PowerShell, the loop is 1..5 | ForEach-Object { curl.exe -s -o NUL -w "%{http_code} " -X POST --data-binary "@data/discharges_night.csv" -H "Content-Type: text/csv" http://127.0.0.1:8000/v1/jobs }.

The output (without the date and server headers):

202 202 202 202 429
HTTP/1.1 429 Too Many Requests
retry-after: 60
content-length: 130
content-type: application/json
x-request-id: 765d8c1fb60f

{"error":{"code":"too_many_jobs","message":"Too many jobs are waiting. Try again later.","request_id":"765d8c1fb60f","fields":[]}}

The first job runs at once and the next three wait. The fifth gets 429 with Retry-After: 60, instead of a queue that grows without limit.

Part 4: fetch the list, and delete it on time

The client

The care team’s system runs scripts/follow_up.py. Open it. The part that matters is the loop in wait:

scripts/follow_up.py

def wait(client: httpx.Client, location: str) -> dict:
    started = time.perf_counter()
    while True:
        response = client.get(location)
        if response.status_code != 200:
            sys.exit(f"{response.status_code} {response.text}")
        job = response.json()
        if job["status"] not in ("queued", "running"):
            return job
        print(f"{time.perf_counter() - started:5.0f} s  {job['status']}: {job['scored']} of {job['records']} scored")
        time.sleep(int(response.headers.get("Retry-After", "5")))

While the job is queued or running, the client sleeps for Retry-After seconds. The server, not the client, decides how often it wants to be asked.

With the service still at 5 ms per row, wait until the four night jobs are done (about a minute), then run the client:

Second terminal

python scripts/follow_up.py data/discharges_night.csv

The output:

Accepted: job 8f6af7074be0, 2801 rows.
    0 s  running: 0 of 2801 scored
    2 s  running: 0 of 2801 scored
    4 s  running: 0 of 2801 scored
    6 s  running: 1000 of 2801 scored
    8 s  running: 1000 of 2801 scored
   10 s  running: 2000 of 2801 scored
   12 s  running: 2000 of 2801 scored
Done in 14.172 s: 2801 scored, 0 rejected, 161 to call (model 1.0.0).
Saved out/8f6af7074be0.csv. The service deletes its copy at 2026-10-10T14:56:29+00:00.

Only 161 of 2,801 stays (5.7%) are on this night’s list, not 10%: the threshold was set on other patients. Adaeze’s team should count the calls each night.

Check the list against what happened

Restart the service without the simulated cost, and score the week. Then join the list with outcomes_week.csv, which says who came back within 30 days (a real team gets this a month later).

Second terminal

python scripts/follow_up.py data/discharges_week.csv

Accepted: job 5d954af1d20b, 19608 rows.
    0 s  running: 0 of 19608 scored
Done in 0.313 s: 19608 scored, 0 rejected, 1866 to call (model 1.0.0).
Saved out/5d954af1d20b.csv. The service deletes its copy at 2026-10-10T14:57:33+00:00.

Use your own job ID in the file name:

Second terminal

python -c "import pandas as pd; f = pd.read_csv('out/5d954af1d20b.csv').merge(pd.read_csv('data/outcomes_week.csv'), on='encounter_id'); c = f[f['call']]; print(len(c), 'calls;', c['readmitted_30'].sum(), 'came back within 30 days: precision', round(c['readmitted_30'].mean(), 3))"

The output:

1866 calls; 504 came back within 30 days: precision 0.27

The list gives exactly the test result of training: 1,866 calls, precision 0.270. This end-to-end check is stronger than the 10 parity cases: it covers the whole path, from the file to the list.

Delete the list on time

Imani’s condition is a setting, KEEP_RESULTS_HOURS (48 by default). To see it work, start the service with 0.001 hours (3.6 seconds):

macOS

KEEP_RESULTS_HOURS=0.001 fastapi dev

Windows

$env:KEEP_RESULTS_HOURS = "0.001"; fastapi dev

Linux

KEEP_RESULTS_HOURS=0.001 fastapi dev

Submit the sample, wait 5 seconds, and ask for the list (use your job ID):

Second terminal

curl -s -X POST --data-binary @data/discharges_sample.csv -H "Content-Type: text/csv" http://127.0.0.1:8000/v1/jobs
curl -si http://127.0.0.1:8000/v1/jobs/2e806736439a/results

The output of the second command (without the date and server headers):

HTTP/1.1 410 Gone
content-length: 131
content-type: application/json
x-request-id: 18ceb6c02883

{"error":{"code":"job_expired","message":"The list expired at 2026-10-08T14:57:13+00:00.","request_id":"18ceb6c02883","fields":[]}}

410 Gone is not 404 Not Found: the client can tell “you waited too long” from “wrong ID”. The worker deletes expired lists when it next wakes, at most a minute later. Stop the service (on Windows, also Remove-Item Env:KEEP_RESULTS_HOURS).

Part 5: a broken file must not stop the worker

Exports go wrong. data/discharges_typos.csv has 3 broken cells in 200 rows. data/discharges_hours.csv has time_in_hospital in hours instead of days, in every row. Start the service (fastapi dev) and send both, then the sample:

Second terminal

python scripts/follow_up.py data/discharges_typos.csv
python scripts/follow_up.py data/discharges_hours.csv
python scripts/follow_up.py data/discharges_sample.csv

The output:

Accepted: job 6dd7d07f8ca2, 200 rows.
    0 s  running: 0 of 200 scored
Done in 0.007 s: 197 scored, 3 rejected, 3 to call (model 1.0.0).
Saved out/6dd7d07f8ca2.csv. The service deletes its copy at 2026-10-10T15:10:14+00:00.
Accepted: job 75b88bf28643, 200 rows.
Job 75b88bf28643 failed: 200 of 200 rows do not match the contract
Accepted: job da9423e215e2, 5 rows.
Done in 0.003 s: 5 scored, 0 rejected, 0 to call (model 1.0.0).
Saved out/da9423e215e2.csv. The service deletes its copy at 2026-10-10T15:10:16+00:00.

Look at the first job’s status (use your job ID):

Second terminal

curl -s http://127.0.0.1:8000/v1/jobs/6dd7d07f8ca2

{"job_id":"6dd7d07f8ca2","status":"done","records":200,"scored":197,"rejected":3,"rejected_lines":[{"line":19,"encounter_id":"1115640","fields":["age_years"]},{"line":66,"encounter_id":"2928240","fields":["time_in_hospital"]},{"line":153,"encounter_id":"5683578","fields":["number_inpatient"]}],"to_call":3,"unknown_values":{},"created_at":"2026-10-08T15:10:14+00:00","finished_at":"2026-10-08T15:10:14+00:00","expires_at":"2026-10-10T15:10:14+00:00","seconds":0.007,"model_version":"1.0.0","results_url":"/v1/jobs/6dd7d07f8ca2/results","error":null}

• A few bad rows are set aside, each with its line and field, and the other 197 are scored.
• A wrong file fails as a whole: more than 5% bad rows, no list, as in batch inference.
• The next job runs: the failed job did not stop the worker.

Part 6: put it in a container, and restart it

Package the service as in Container resource settings. The starter has requirements-run.txt. Make three files. The Dockerfile:

Dockerfile

FROM python:3.14.8-slim-trixie

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1 \
    PIP_NO_CACHE_DIR=1 \
    PIP_DISABLE_PIP_VERSION_CHECK=1 \
    PIP_ROOT_USER_ACTION=ignore

RUN useradd --create-home --uid 10001 app

WORKDIR /app

COPY requirements-run.txt .
RUN pip install -r requirements-run.txt

COPY bundles ./bundles
COPY pyproject.toml .
COPY readmission ./readmission

USER app

EXPOSE 8000
CMD ["fastapi", "run", "--host", "0.0.0.0", "--port", "8000"]

.dockerignore, so that the data and the virtual environment stay out of the image:

.dockerignore

# Send only what the image needs. Data, tests, jobs, .env and .venv stay out.
*
!requirements-run.txt
!pyproject.toml
!readmission/
!bundles/
**/__pycache__/

And compose.yaml:

compose.yaml

# The readmission service on this computer, with the limits of a small cloud
# container. MODEL_SHA256 comes from .env (copy .env.example).
services:
  readmission:
    image: readmission-service:1.0.0
    build: .
    ports:
      - "127.0.0.1:8000:8000"
    environment:
      MODEL_BUNDLE: bundles/readmission-1.0.0
      MODEL_SHA256: ${MODEL_SHA256:?Set MODEL_SHA256 in .env}
      EXTRA_MS_PER_RECORD: ${EXTRA_MS_PER_RECORD:-0}  # SIMULATED slow model; 0 in production
    cpus: 1.0
    mem_limit: 512m
    memswap_limit: 512m
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/ready')"]
      interval: 5s
      timeout: 3s
      retries: 3
      start_period: 10s

Copy .env.example to .env (Windows: Copy-Item .env.example .env), make sure that no fastapi dev runs on port 8000, and start the container:

docker compose up -d --build --wait
docker compose logs

The output of docker compose logs (the last lines):

readmission-1  | INFO:     Started server process [1]
readmission-1  | INFO:     Waiting for application startup.
readmission-1  | INFO readmission: model 1.0.0 loaded (bundle 89056e4cfab4...) in 874 ms
readmission-1  | INFO:     Application startup complete.
readmission-1  | INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
readmission-1  | INFO readmission: GET /ready 200 23.5ms id=89af3e04814b

The image is 787 MB on disk (166 MB compressed), like the course’s service. In the container (1 CPU), the week job took 0.33 to 0.73 seconds with the real model.

Now simulate the slow model in the container: add the line EXTRA_MS_PER_RECORD=5 to .env and run docker compose up -d --wait again. Submit the week, and ask for its status after about 10 seconds (use your job ID):

curl -s -X POST --data-binary @data/discharges_week.csv -H "Content-Type: text/csv" http://127.0.0.1:8000/v1/jobs
curl -s http://127.0.0.1:8000/v1/jobs/8830f6b92eff

{"job_id":"8830f6b92eff","status":"running","records":19608,"scored":2000,"rejected":0,"rejected_lines":[],"to_call":null,"unknown_values":{},"created_at":"2026-10-08T14:58:37+00:00","finished_at":null,"expires_at":null,"seconds":null,"model_version":null,"results_url":null,"error":null}

Now do what a platform does when it moves a container, applies an update or recovers from a crash: restart it.

Common mistake

After a restart, the job is gone

Symptom. You restart the container, then ask for the job that was running:

docker compose restart
curl -s http://127.0.0.1:8000/v1/jobs/8830f6b92eff

 Container c17cs3-readmission-1 Restarting
 Container c17cs3-readmission-1 Started
{"error":{"code":"job_not_found","message":"There is no job with this ID.","request_id":"e0e43350b394","fields":[]}}

The old process’s log ends calmly: Application shutdown complete. (Our container name starts with c17cs3-, our test project name; yours is readmission-service-readmission-1.)

Cause. MemoryStore keeps the jobs, the files and the progress in one process’s memory. A restart ends the process, and its memory with it.

Fix. Keep the jobs and the lists outside the process, in storage that outlives it, and let the new process find the jobs that were running. That is Part 7.

What it means for a real AI service. This failure is quiet: /ready answers 200, the logs have no error, and the caller sees 404 for a job that the service accepted with 202. At 07:30 the nurses have no list, and nothing alerted anyone. A 202 is a promise; a service that keeps it only while nothing restarts does not keep it.

Part 7: keep jobs where a restart cannot reach them

The idea

In a cloakroom, an attendant who only remembers the hooks loses everything at a change of shift. A book at the counter lets any attendant find any coat. The book is durable storage: it outlives the person who wrote in it.

For the service, the book is SQLite, a whole database in one file, included in Python. It and the files and lists live on a Docker volume, as in Persistent data. At start-up, a job that was running goes back to the queue and is scored again from the first row.

The same restart, two outcomes. In memory, the job disappears. In SQLite on a volume, it goes back to the queue and finishes.

Text description of the diagram

The restart costs the time of the job that was running, not the job. That is safe here for two reasons. The same file with the same bundle always gives the same list. And the worker writes the list in one step, at the end.

Step 1: the SQLite store

Make a new file, readmission/store.py. It has the same methods as MemoryStore, so the runner does not change:

readmission/store.py

"""Jobs that survive a restart: the jobs in SQLite, the files and lists on disk.

    jobs/jobs.db                 one row per job (SQLite, a database in one file)
    jobs/<id>.csv                the discharge file, until the job has run
    jobs/<id>.list.csv           the follow-up list, until the job expires

Put jobs/ on a volume, so that it outlives the container. At start-up,
recover() puts a job that was running back in the queue: a restart then costs
the time of that job, not the job.

One service process only: two processes would each start a worker, and each
would recover the other's running job. More processes need a queue service.
"""

import json
import os
import sqlite3
from contextlib import contextmanager
from dataclasses import asdict
from pathlib import Path

from readmission.jobs import Job, now

SCHEMA = """CREATE TABLE IF NOT EXISTS jobs (
    id TEXT PRIMARY KEY,
    status TEXT NOT NULL,
    created_at TEXT NOT NULL,
    expires_at TEXT,
    job TEXT NOT NULL
)"""


class SqliteStore:
    def __init__(self, folder: Path):
        self.folder = Path(folder)
        self.folder.mkdir(parents=True, exist_ok=True)
        self.db = self.folder / "jobs.db"
        with self.connect() as db:
            db.execute(SCHEMA)

    @contextmanager
    def connect(self):
        """One connection per call, safe across threads: commit at the end, then close."""
        db = sqlite3.connect(self.db, timeout=10)
        try:
            with db:
                yield db
        finally:
            db.close()

    def write(self, db: sqlite3.Connection, job: Job) -> None:
        db.execute(
            "INSERT OR REPLACE INTO jobs VALUES (?, ?, ?, ?, ?)",
            (job.id, job.status, job.created_at, job.expires_at, json.dumps(asdict(job))),
        )

    def add(self, job: Job, data: bytes) -> None:
        (self.folder / f"{job.id}.csv").write_bytes(data)  # the file first, then the row
        with self.connect() as db:
            self.write(db, job)

    def get(self, job_id: str) -> Job | None:
        with self.connect() as db:
            row = db.execute("SELECT job FROM jobs WHERE id = ?", (job_id,)).fetchone()
        return Job(**json.loads(row[0])) if row else None

    def save(self, job: Job) -> None:
        with self.connect() as db:
            self.write(db, job)

    def queued(self) -> int:
        with self.connect() as db:
            return db.execute("SELECT COUNT(*) FROM jobs WHERE status = 'queued'").fetchone()[0]

    def claim_next(self) -> Job | None:
        with self.connect() as db:
            row = db.execute(
                "SELECT job FROM jobs WHERE status = 'queued' ORDER BY created_at, rowid LIMIT 1"
            ).fetchone()
            if row is None:
                return None
            job = Job(**json.loads(row[0]))
            job.status, job.started_at = "running", now()
            self.write(db, job)
        return job

    def file(self, job_id: str) -> bytes:
        return (self.folder / f"{job_id}.csv").read_bytes()

    def finish(self, job: Job, follow_up_list: str | None) -> None:
        if follow_up_list is not None:
            partial = self.folder / f"{job.id}.list.csv.partial"
            partial.write_text(follow_up_list, encoding="utf-8")
            os.replace(partial, self.folder / f"{job.id}.list.csv")  # all or nothing
        with self.connect() as db:
            self.write(db, job)
        (self.folder / f"{job.id}.csv").unlink(missing_ok=True)

    def follow_up_list(self, job_id: str) -> str | None:
        path = self.folder / f"{job_id}.list.csv"
        return path.read_text(encoding="utf-8") if path.exists() else None

    def expire(self) -> int:
        with self.connect() as db:
            rows = db.execute(
                "SELECT job FROM jobs WHERE status != 'expired' AND expires_at <= ?", (now(),)
            ).fetchall()
            for (text,) in rows:
                job = Job(**json.loads(text))
                job.status = "expired"
                self.write(db, job)
                (self.folder / f"{job.id}.list.csv").unlink(missing_ok=True)
        return len(rows)

    def recover(self) -> int:
        """Put the jobs that were running when the service stopped back in the queue."""
        with self.connect() as db:
            rows = db.execute("SELECT job FROM jobs WHERE status = 'running'").fetchall()
            for (text,) in rows:
                job = Job(**json.loads(text))
                job.status, job.started_at, job.scored = "queued", None, 0
                self.write(db, job)
        return len(rows)

Two methods do most of the work. finish writes the list to a .partial file and renames it, so a restart never leaves half a list (the atomic write of a batch job). recover runs at start-up and changes running back to queued.

Step 2: choose the store

In readmission/settings.py, add two fields at the end of the Settings class:

readmission/settings.py

    job_store: str = "sqlite"
    jobs_dir: Path = Path("jobs")

In load_settings, add two lines before the closing parenthesis of Settings(...), and one check after it:

readmission/settings.py

        job_store=env.get("JOB_STORE", "sqlite"),
        jobs_dir=Path(env.get("JOBS_DIR", "jobs")),
    )
    if settings.job_store not in ("sqlite", "memory"):
        raise SettingsError("JOB_STORE must be sqlite or memory")

In readmission/main.py, add the import from readmission.store import SqliteStore, and replace the JobRunner(...) call in lifespan with this:

readmission/main.py

        if settings.job_store == "memory":
            store = MemoryStore()
            logger.warning("jobs are kept in memory: a restart loses them")
        else:
            store = SqliteStore(settings.jobs_dir)
            logger.info("jobs are kept in %s; %d running job(s) queued again", settings.jobs_dir, store.recover())
        app.state.jobs = JobRunner(
            store,
            lambda: app.state.bundle,
            settings.max_queued_jobs,
            settings.keep_results_hours,
            settings.extra_ms_per_record,
        )

memory stays as an option for tests, with a warning in the log.

Step 3: the folder and the volume

The container runs as the user app, which cannot create folders in /app. In the Dockerfile, add these lines before USER app:

Dockerfile

# The job store. A volume mounted here keeps the jobs when the container goes.
RUN mkdir /app/jobs && chown app /app/jobs

In compose.yaml, add the two settings to environment, the volume to the service, and the volume’s name at the end of the file:

compose.yaml

      JOB_STORE: ${JOB_STORE:-sqlite}
      JOBS_DIR: /app/jobs
...
    volumes:
      - jobs:/app/jobs  # the jobs and lists outlive the container
...
volumes:
  jobs:

The finished compose.yaml in the finished project shows where each line goes.

Step 4: prove it

Build and start the new image, with EXTRA_MS_PER_RECORD=5 still in .env:

docker compose up -d --build --wait
docker compose logs | grep "readmission:"

readmission-1  | INFO readmission: model 1.0.0 loaded (bundle 89056e4cfab4...) in 889 ms
readmission-1  | INFO readmission: jobs are kept in /app/jobs; 0 running job(s) queued again
readmission-1  | WARNING readmission: SIMULATED slow model: 5.0 ms more per row

On Windows, use docker compose logs | Select-String "readmission:". Submit the week, wait about 10 seconds, restart, and ask again (use your job ID):

curl -s -X POST --data-binary @data/discharges_week.csv -H "Content-Type: text/csv" http://127.0.0.1:8000/v1/jobs
curl -s http://127.0.0.1:8000/v1/jobs/66eaf210964f
docker compose restart
curl -s http://127.0.0.1:8000/v1/jobs/66eaf210964f

The output of the two status requests, before and after the restart:

{"job_id":"66eaf210964f","status":"running","records":19608,"scored":2000,"rejected":0,"rejected_lines":[],"to_call":null,"unknown_values":{},"created_at":"2026-10-08T14:59:25+00:00","finished_at":null,"expires_at":null,"seconds":null,"model_version":null,"results_url":null,"error":null}
{"job_id":"66eaf210964f","status":"running","records":19608,"scored":0,"rejected":0,"rejected_lines":[],"to_call":null,"unknown_values":{},"created_at":"2026-10-08T14:59:25+00:00","finished_at":null,"expires_at":null,"seconds":null,"model_version":null,"results_url":null,"error":null}

The log of the new process says INFO readmission: jobs are kept in /app/jobs; 1 running job(s) queued again. Fetch the list with the client, by the job’s ID:

python scripts/follow_up.py --job 66eaf210964f

    0 s  running: 1000 of 19608 scored
    2 s  running: 1000 of 19608 scored
    4 s  running: 2000 of 19608 scored
...
   91 s  running: 19000 of 19608 scored
Done in 99.276 s: 19608 scored, 0 rejected, 1866 to call (model 1.0.0).
Saved out/66eaf210964f.csv. The service deletes its copy at 2026-10-10T15:01:19+00:00.

The job survived the restart; its progress did not (scored went back from 2,000 to 0). That is the price of this simple design.

A real deployment does more than a restart: it makes a new container from the image. Try that too:

docker compose down
docker compose up -d --wait
curl -s http://127.0.0.1:8000/v1/jobs/66eaf210964f/results
docker compose exec readmission ls -la /app/jobs

rank,encounter_id,score,call
1,291808098,0.876674,true
2,33695262,0.818332,true
...
-rw-r--r-- 1 app  app  588381 Oct  8 15:01 66eaf210964f.list.csv
-rw-r--r-- 1 app  app   12288 Oct  8 15:01 jobs.db

The new container found the list on the volume. The input file was deleted when the job ended, as Imani asked.

The three places for the jobs, measured:

Where the jobs are	After a restart	After a new container
In the process’s memory (MemoryStore)	Lost (404)	Lost (404)
SQLite in the container’s own files, no volume	Kept (200)	Lost (404)
SQLite on a volume	Kept, a running job is queued again	Kept

When this design is enough. It is enough for one process with one worker thread. With more processes, each would start a worker and put the other’s running job back in the queue. More processes need a queue service, for example Azure Queue Storage or Amazon SQS. It gives each job to one worker at a time. The lists then go to object storage. The course Scaling APIs and AI Workloads covers both. The worker also shares the CPU with the API: during the week job, one status request took 81 ms instead of under 1 ms.

Part 8: test it

The tests make every rule of this page repeatable. Make tests/test_jobs.py:

tests/test_jobs.py

import time
from pathlib import Path

from fastapi.testclient import TestClient

from readmission.jobs import Job
from readmission.store import SqliteStore
from tests.conftest import CSV, SAMPLE


def wait(client, location, until=("done", "failed")):
    for _ in range(200):
        body = client.get(location).json()
        if body["status"] in until:
            return body
        time.sleep(0.02)
    raise AssertionError(f"still {body['status']}")


def new_client():
    from readmission.main import create_app

    return TestClient(create_app())


def test_a_job_is_accepted_then_done(client):
    response = client.post("/v1/jobs", content=SAMPLE, headers=CSV)
    assert response.status_code == 202
    assert response.headers["Retry-After"] == "2"
    assert response.json()["status"] == "queued"
    body = wait(client, response.headers["Location"])
    assert body["status"] == "done" and body["records"] == body["scored"] == 5
    follow_up = client.get(body["results_url"])
    assert follow_up.headers["content-type"].startswith("text/csv")
    lines = follow_up.text.splitlines()
    assert lines[0] == "rank,encounter_id,score,call" and len(lines) == 6


def test_a_file_must_be_csv(client):
    response = client.post("/v1/jobs", json={"rows": []})
    assert response.status_code == 415
    assert response.json()["error"]["code"] == "not_csv"


def test_a_sensitive_column_is_refused_before_any_job(client):
    header, first = SAMPLE.decode().splitlines()[:2]
    response = client.post("/v1/jobs", content=f"{header},race\n{first},Caucasian\n", headers=CSV)
    assert response.status_code == 422
    assert response.json()["error"]["fields"] == ["race"]


def test_a_large_file_is_413(monkeypatch):
    monkeypatch.setenv("MAX_UPLOAD_BYTES", "500")
    with new_client() as client:
        response = client.post("/v1/jobs", content=SAMPLE, headers=CSV)
    assert response.status_code == 413
    assert response.json()["error"]["code"] == "file_too_large"


def test_a_full_queue_is_429(monkeypatch):
    monkeypatch.setenv("EXTRA_MS_PER_RECORD", "100")  # simulated: each job takes 0.5 s
    monkeypatch.setenv("MAX_QUEUED_JOBS", "1")
    with new_client() as client:
        codes = [client.post("/v1/jobs", content=SAMPLE, headers=CSV).status_code for _ in range(4)]
        assert codes[0] == 202 and codes[-1] == 429
        assert client.post("/v1/jobs", content=SAMPLE, headers=CSV).headers["Retry-After"] == "60"


def test_a_few_bad_rows_are_rejected_and_the_rest_scored(client):
    response = client.post("/v1/jobs", content=Path("data/discharges_typos.csv").read_bytes(), headers=CSV)
    body = wait(client, response.headers["Location"])
    assert body["status"] == "done" and body["scored"] == 197 and body["rejected"] == 3
    assert body["rejected_lines"][0] == {"line": 19, "encounter_id": "1115640", "fields": ["age_years"]}


def test_a_failed_job_does_not_stop_the_worker(client):
    hours = client.post("/v1/jobs", content=Path("data/discharges_hours.csv").read_bytes(), headers=CSV)
    sample = client.post("/v1/jobs", content=SAMPLE, headers=CSV)
    failed = wait(client, hours.headers["Location"])
    assert failed["status"] == "failed" and failed["results_url"] is None
    assert "rows do not match the contract" in failed["error"]
    assert client.get(f"{hours.headers['Location']}/results").json()["error"]["code"] == "job_failed"
    assert wait(client, sample.headers["Location"])["status"] == "done"


def test_results_before_the_end_is_409(monkeypatch):
    monkeypatch.setenv("EXTRA_MS_PER_RECORD", "100")
    with new_client() as client:
        location = client.post("/v1/jobs", content=SAMPLE, headers=CSV).headers["Location"]
        response = client.get(f"{location}/results")
        assert response.status_code == 409 and response.headers["Retry-After"] == "2"


def test_an_unknown_job_is_404(client):
    assert client.get("/v1/jobs/nothere").json()["error"]["code"] == "job_not_found"


def test_a_list_expires(monkeypatch):
    monkeypatch.setenv("KEEP_RESULTS_HOURS", "0.0003")  # about 1 second
    with new_client() as client:
        location = client.post("/v1/jobs", content=SAMPLE, headers=CSV).headers["Location"]
        assert wait(client, location)["status"] == "done"
        time.sleep(1.2)
        assert client.get(location).json()["status"] == "expired"
        assert client.get(f"{location}/results").status_code == 410


def test_a_memory_job_is_lost_on_restart(monkeypatch):
    monkeypatch.setenv("JOB_STORE", "memory")
    with new_client() as client:
        location = client.post("/v1/jobs", content=SAMPLE, headers=CSV).headers["Location"]
        assert wait(client, location)["status"] == "done"
    with new_client() as client:  # a restart
        assert client.get(location).status_code == 404


def test_a_stored_job_survives_a_restart():
    with new_client() as client:
        location = client.post("/v1/jobs", content=SAMPLE, headers=CSV).headers["Location"]
        assert wait(client, location)["status"] == "done"
    with new_client() as client:  # a restart
        assert client.get(location).json()["status"] == "done"
        assert len(client.get(f"{location}/results").text.splitlines()) == 6


def test_a_running_job_is_queued_again_after_a_restart(tmp_path):
    store = SqliteStore(tmp_path / "jobs")  # the JOBS_DIR of the test settings
    store.add(Job(id="interrupted1", records=5, input_sha256="-"), SAMPLE)
    assert store.claim_next().status == "running"  # the service stops here
    with new_client() as client:
        assert wait(client, "/v1/jobs/interrupted1")["status"] == "done"

In a test, a “restart” is a second app object with the same settings: it sees only what the store kept. The last test leaves a job running, as a crash would, and checks that the next start finishes it.

python -m pytest

The output (last lines):

tests/test_bundle.py ......                                              [ 18%]
tests/test_contract.py .........                                         [ 46%]
tests/test_health.py ....                                                [ 59%]
tests/test_jobs.py .............                                         [100%]

============================== 32 passed in 2.65s ==============================

The 13 job tests cover each code that a caller can see, each state of a job, the 5% rule and the restart.

Our run-through

We ran every step of this page from starter.zip in a new folder with a new virtual environment. The commands and full outputs are in the course’s repository (reference/c17/case-studies/readmissions-async/RUNS.md).

Check	Result
Tests	19 passed at the start; 32 at the end; the finished project: 32 passed
Training	Byte for byte the same model.joblib as the bundle (a7fbea052900…)
Real model, week file	0.19 to 0.22 s in one request; 0.31 to 0.36 s as a job; 0.33 to 0.73 s in the container
Simulated 5 ms per row	The request timed out at 30 s; the job finished in 99 s
Queue, retention, bad files	429 for the fifth job; 410 after the retention time; 3 rows rejected, a wrong file failed
Restart	In memory: the job lost (404). In SQLite on a volume: queued again, then done; the list kept in a new container
Not tested	Windows and Linux; more than one process; a cloud deployment

Responsible use

A follow-up list looks harmless: it only says whom to call. It is not harmless, because people act on lists, and a list can hide who is missing from it. Imani’s conditions are part of the product.

The list never decides on its own. It ranks stays for an offer of a call. It does not decide who may leave hospital, who gets less care, or what anyone pays. A low score does not mean “safe”: the calls reach only 22% of the patients who come back, and about half of those who come back had no hospital stay in the year before, so the model scores most of them low. The nurses must be able to add any patient to the list, and the team must say so in its procedures.

The service takes only what it needs. The contract refuses any column outside the 19 features, so the service rejects a file with race, gender or the payer before a job exists (the test test_a_sensitive_column_is_refused_before_any_job). The files have no names and no patient numbers. The service deletes the input file when its job ends, and the list after 48 hours, and the log has job IDs and counts, never a row.

The data is old and from one country. The model learned from US hospitals between 1999 and 2008, with ICD-9 codes. Care, coding and insurance have changed since. A hospital must train and test it on its own, recent patients before any real use. The machine learning case study also found that it works less well for the oldest patients. Check precision and recall by age, race, gender and payer every month, with the outcomes when they arrive.

The signal is weak. A ROC AUC of 0.663 is far from 1. The list is better than chance: a precision of 0.27 against 0.11. It is only a little better than a simple rule of thumb: call the patients with the most hospital stays in the past year. On the same patients, the machine learning case study measured 0.264 for the top 10% of the model’s list, and 0.247 for the rule. It is not a prediction for one patient.

What a real service would also need. Health data falls under laws such as HIPAA in the United States and the GDPR in Europe (this is not legal advice). A real deployment needs access control for each user, an audit log of who fetched which list, encryption of the volume, and a review by the hospital’s information governance team. This case study builds none of these; it builds the habits that make them possible.

The simulated cost is not a model. EXTRA_MS_PER_RECORD shows when the pattern breaks. It does not show how a real text model would behave: measure that model before you choose its limits.

Make it a portfolio piece

This project shows that you can serve a model whose work does not fit in a request. It also shows that you test what happens when things restart. Publish it as a repository, under your name.

What to put in the repository:

readmission-service/
  README.md             what it is, how to run it, the endpoints, the data and its licence
  MODEL_CARD.md         intended use, not for, data, results, limits
  LICENSE               MIT, for the code
  readmission/          contract, bundle checks, scoring, jobs (memory and SQLite), routes
  bundles/              readmission-1.0.0 with SHA256SUMS
  data/                 the discharge files and data/README.md (CC BY 4.0, with the changes)
  training/train.py     makes the same model file again from the public data
  tests/                32 tests, with the restart tests
  Dockerfile, compose.yaml, .env.example
  scripts/follow_up.py  the client that respects Retry-After

1. Start from your project folder, or from the finished project. Check that .gitignore lists .venv/, .env, jobs/, out/ and data/raw/.
2. Make the repository: git init, then git add . and git commit -m "Readmission scoring service with queued jobs".
3. Check with git ls-files that the bundle and data/README.md are in it, and that no .env file is.
4. Push it to GitHub as a public repository. Keep LICENSE (MIT) for the code; the data keeps its own licence (CC BY 4.0), which the README states with the attribution.
5. Rewrite the README’s first paragraph in your words, and add your own measurements: the job times, and the restart test.

A README template:

# Readmission scoring service

A FastAPI service that scores hospital discharge files as queued jobs and returns
a follow-up list for a care team. It supports calls; it never decides on its own.

- Model: logistic regression on public US hospital data (1999-2008, CC BY 4.0);
  test precision 0.27 at a 10% call capacity (0.11 at random).
- Jobs: 202 Accepted with Location and Retry-After; progress while running;
  429 when 3 jobs wait; 413 and 422 before a job exists; lists deleted after 48 h.
- Durability: jobs in SQLite on a Docker volume; a running job is queued again
  after a restart (tested).
- Measured: the real model scores 19,608 stays in 0.36 s; a simulated 5 ms per
  row takes 99 s, beyond a 30 s gateway limit.

How to describe it in an interview: “I served a readmission model as an asynchronous job API. The caller gets 202 with a job address, polls with Retry-After and fetches the list. The service has a queue limit, partial-failure rules and a retention time. In-memory jobs disappeared on a container restart, so I moved them to SQLite on a volume, with a test that proves a job survives a restart.”

Go further:

1. An idempotency key. If Rafael’s export loses the 202 answer and sends the file again, the service makes two jobs. Accept an Idempotency-Key header, and return the existing job for a key that it has seen.
2. A callback instead of polling. Let the client give a URL, and call it when the job is done. Keep polling as the fallback.
3. A separate worker process. Move the worker into its own container that reads jobs from the same store, and measure the status latency while a large job runs.
4. Reasons for each call. Logistic regression can split each score into the contributions of the features (the machine learning case study shows how). Add the two largest to each row of the list, so the nurse can open the call with them.
5. A monthly slice report. Join the lists with the outcomes, and report precision and recall by age band, as in the machine learning case study.

Knowledge check

Check your understanding

Rafael says: the real model scores the week in 0.2 seconds, so the job is a waste. When is he right, and what would make you keep the job anyway?

Show the answer

He is right for today’s model and files. A synchronous request that takes 0.2 seconds is simpler, and a job adds a store, polling, retention and more failure states. Keep the job if the work times the size can exceed the caller’s budget. The measured week file passes the 30-second gateway limit at about 1.5 ms more per row, and the team plans a slower model. Keep it also because the caller is an unattended program at 02:00. With a job, the list survives a restart and waits until 07:30. A synchronous answer disappears with the connection. The decision rests on measurements, so record them and measure again when the model changes.

Check your understanding

At start-up, recover() puts a running job back in the queue. Why is it safe to score that file again from the first row, and what would make it unsafe?

Show the answer

It is safe because the job is repeatable. The store keeps the input file until the job ends, and the digest fixes the bundle. The same rows with the same bundle always give the same scores. The worker also writes the list in one step (a .partial file, then a rename), so a half-finished run leaves no half list. It would be unsafe if the job had side effects that must not happen twice, for example a message to each patient. It would also be unsafe if the model or the input could change between the two runs. Then the job would need to record what it already did.

Check your understanding

Rafael wants more throughput and starts the service with two worker processes, fastapi run --workers 2, with the SQLite store. Predict two things that go wrong.

Show the answer

Each process starts its own worker thread. First, each process runs recover() at start-up. So a process that restarts puts the other process’s running job back in the queue, and the job runs twice, at the same time. Second, both threads can read the same oldest queued job before either writes running, so two workers may take one job. The queue limit is still shared, because both processes count the same table. For this design, the fix is one process with one worker. For more processes, use a queue service: it gives each job to exactly one worker, and to another only when that worker stops answering.

Recap

• A job is worth its cost only when the work times the size can exceed the caller’s budget. Measure first. The real model scored 19,608 stays in about 0.2 seconds. A simulated 5 ms per row took 99 seconds, past a 30-second gateway.
• The request does only fast checks (type, size, columns, rows, queue) and answers 202 Accepted with Location and Retry-After. The worker does the scoring, in chunks, with progress.
• A full queue answers 429 with Retry-After. A few bad rows are set aside by line; more than 5% fail the job; a failed job never stops the worker.
• A list is kept for a stated time, then deleted: 410 Gone tells the client why.
• In-memory jobs disappear on a restart, quietly. Jobs in SQLite on a volume survive a restart and a new container, and a running job is queued again.
• One process, one worker: more processes need a queue service and shared storage.
• The list supports a care team’s calls. It never decides about a patient on its own, and it takes only the columns it needs. Its data is old: retrain and check it on your own patients first.

Clean up. Stop the container and delete its volume, which also deletes the jobs and lists: docker compose down -v. Remove the image: docker image rm readmission-service:1.0.0.

Next: Final assignment