Prédire les réadmissions à l'hôpital de façon asynchrone

Ce que vous saurez faire

Mettez en service un modèle dont le travail ne tient pas dans une seule requête. Un modèle de réadmission hospitalière évalue un gros fichier de sorties comme tâche en file d’attente. Le client reçoit une réponse immédiate, interroge le service plus tard et récupère la liste de suivi quand elle est prête. Ensuite, vous faites en sorte que les tâches survivent à un redémarrage. Un conteneur redémarre plus souvent que vous le pensez.

Étude de casPièce de portfolio

Jeu de données

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

Licence

CC BY 4.0 (données); MIT (code)

Durée

Environ 3 heures

S’exécute dans

VS Code, Python 3.12+

Compétences utilisées

• Tâches asynchrones
• 202 Accepted et interrogations périodiques
• Limites de file et de taille
• Échec partiel
• Conservation des résultats
• Stockage durable des tâches
• Conteneurs et volumes

Télécharger le projet de départ (.zip)Projet terminé (.zip)

Décompressez le projet de départ, ouvrez le dossier dans VS Code et suivez les étapes ci-dessous. Elles commencent par un nouvel environnement virtuel.

Le projet de départ est un petit service avec un paquet de modèle vérifié et un premier point de terminaison synchrone. Vous remplacez ce point de terminaison par des tâches et placez le service dans un conteneur. Vous le faites échouer avec un redémarrage, puis vous le corrigez.

Avant de commencerCe qu’il faut savoir et avoir sous la mainEnviron 180 minutes

Ce que vous devez savoir

L’Inférence asynchrone joue le rôle principal. Voici les autres leçons que cette étude de cas utilise le plus :

• Le paquet, son empreinte et ses six barrières, dans Confiance et intégrité des artefacts.
• Le travail au démarrage et l’état de préparation, dans Charger et initialiser.
• Le Dockerfile, Compose et les limites de ressources, dans Paramètres de ressources du conteneur.
• Les volumes, dans le cours sur Docker : Données persistantes.

Vous n’avez pas besoin du cours d’apprentissage automatique. Si vous avez fait son étude de cas Choisir un modèle pour les réadmissions à l’hôpital, voici le modèle retenu. Maintenant, vous le mettez en service.

Ce qu’il vous faut sur votre ordinateur

• Python 3.12 ou une version plus récente, VS Code avec l’extension Python, et curl.
• Docker Desktop, comme dans Paramètres de ressources du conteneur.
• Environ 1,2 Go d’espace disque libre : 360 Mo pour l’environnement virtuel et 790 Mo pour l’image.

Testé avec Python 3.14.6 et Docker Desktop (Docker Engine 29.6.1) sur macOS, avec une puce Apple. Les paquets : FastAPI 0.142.2, scikit-learn 1.9.1, pandas 3.0.6, NumPy 2.5.3, joblib 1.6.0 et pytest 9.1.1. Les étapes pour Windows et Linux ne sont pas testées.

La situation

Riverbend Health exploite un réseau d’hôpitaux. Son équipe de coordination des soins, dirigée par Adaeze Obi, téléphone aux patients dans les premiers jours après leur sortie. Une infirmière vérifie les médicaments, réserve une visite de suivi et demande qui aide à la maison. L’équipe peut appeler environ un patient sorti sur dix. Elle a donc besoin d’une liste : qui appeler en premier?

Un modèle évalue déjà chaque séjour hospitalier pour prédire le risque de réadmission dans les 30 jours. Il vient de l’étude de cas du cours d’apprentissage automatique. Rafael Costa, un ingénieur de données, exporte les sorties chaque nuit à 2 h dans un seul fichier CSV. Les infirmières commencent à 7 h 30. Jun Park, un développeur, crée le service qui relie les deux. Vous êtes Jun.

Imani Brooks, responsable de la sécurité clinique, fixe trois conditions. La liste indique seulement à qui proposer un appel en premier : elle ne décide jamais seule quoi que ce soit pour un patient. Le service reçoit seulement les colonnes dont le modèle a besoin. Et aucune liste n’est conservée plus de 48 heures.

Rafael demande : « Mon exportation peut envoyer le fichier et attendre la réponse. Pourquoi aurait-on besoin d’autre chose? » La partie 2 répond avec des mesures.

Ce que vous allez créer

Un service avec trois points de terminaison pour les tâches, selon les conventions du cours. Il comprend un paquet vérifié, un format d’erreur unique, une clé d’API, /health et /ready, et un conteneur avec des limites.

Point de terminaison	Son rôle
POST /v1/jobs	Accepter un fichier de sorties (CSV) comme tâche : 202 Accepted (accepté) immédiatement, avec l’adresse de la tâche
GET /v1/jobs/{id}	L’état de la tâche : en file d’attente, en cours (avec progression), terminée, échouée ou expirée
GET /v1/jobs/{id}/results	La liste de suivi (CSV), avec le score le plus élevé en premier, pendant 48 heures
GET /v1/model	La version du modèle, l’empreinte de son paquet et son seuil

Chaque partie commence par un problème : un fichier lent, une rafale de fichiers, une exportation défectueuse, une liste conservée trop longtemps, un redémarrage.

Le modèle et ses données

Les données proviennent du jeu de données Diabetes 130-US Hospitals for Years 1999–2008. Il est de Clore, Cios, DeShazo et Strack, 2014; UCI Machine Learning Repository, doi:10.24432/C5230J, sous licence CC BY 4.0. Il contient 101 766 séjours de patients diabétiques dans 130 hôpitaux américains. Les données sont dépersonnalisées, mais elles concernent la santé de vraies personnes : ne tentez jamais de découvrir leur identité. Riverbend Health et son équipe sont inventés.

Le modèle est celui retenu dans l’étude de cas d’apprentissage automatique. C’est une régression logistique utilisant 19 variables connues le jour de la sortie. Par exemple : le mode d’arrivée, la destination suivante, le groupe du diagnostic principal et les médicaments. La race, le genre et le payeur ne sont pas des entrées. Le modèle a appris à partir de 59 556 séjours de 41 999 patients, répartis par patient.

Le seuil est de 0,18 : le score qui délimite les 10 % des séjours de validation les mieux classés. Il correspond à la capacité de l’équipe : un appel sur dix.

Données	Aire sous la courbe ROC	Appels (score ≥ 0,18)	Précision des appels	Rappel
Validation, 20 179 séjours	0,671	2 013 (10,0 %)	0,261	0,231
Test, 19 608 séjours (utilisées une seule fois)	0,663	1 866 (9,5 %)	0,270	0,221

Environ 27 appels sur 100 joignent un patient qui revient à l’hôpital (au hasard, environ 11). Mais ils ne joignent que 22 % de ces patients : la liste aide les infirmières, elle ne les remplace pas. (L’étude de cas d’apprentissage automatique rapporte 1 870 appels de test avec le seuil non arrondi.)

Les fichiers de sorties contiennent les 19 608 séjours de test. Un fichier représente une semaine (tous les séjours, 2,8 Mo), l’autre une nuit (le premier septième, 2 801 séjours). Les données n’ont pas de dates : « semaine » et « nuit » servent donc à l’enseignement. Un fichier contient le encounter_id (identifiant du séjour) et les 19 variables : aucun nom, aucun numéro de patient.

Préparer le projet de départ

1. Téléchargez le projet de départ avec le bouton dans l’encadré en haut de cette page. Décompressez-le dans votre dossier projects. Vous obtenez un dossier readmission-service.

2. Dans VS Code, ouvrez le dossier : File › Open Folder… (fichier › ouvrir un dossier), puis choisissez readmission-service.

3. Ouvrez un terminal dans VS Code : Terminal › New Terminal (terminal › nouveau terminal). Il démarre dans le dossier du projet.

4. Créez et activez un environnement virtuel :

   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

   L’invite de commande commence par (.venv).

5. Installez les versions exactes des paquets :

   python -m pip install -r requirements.txt

   Cela prend une ou deux minutes. L’environnement utilise environ 360 Mo.

6. Indiquez à VS Code d’utiliser cet environnement. Ouvrez la palette de commandes (Ctrl+Shift+P, ou Cmd+Shift+P sur macOS), choisissez Python : Select Interpreter (Python : sélectionner l’interpréteur), puis celui dans .venv.

7. Exécutez les tests :

   python -m pytest

La sortie (dernières lignes) :

collected 19 items

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

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

La première exécution est lente parce que Python compile les paquets; les suivantes prennent environ 2 secondes.

Chemin	Contenu
bundles/readmission-1.0.0/	Le paquet du modèle : model.joblib (12 Ko), contract.json, metadata.json, parity_cases.csv (10 cas), SHA256SUMS
readmission/	contract.py (les 19 variables et leurs règles), bundle.py (les six barrières), scoring.py (vérifier et évaluer un fichier), settings.py, errors.py, security.py, main.py, routes.py
data/	Les fichiers de sorties, deux fichiers défectueux et outcomes_week.csv (les résultats réels, pour les vérifications seulement); data/README.md explique leur création
training/train.py, make_bundle.py	Comment le modèle et le paquet ont été créés
scripts/follow_up.py	Le client de l’équipe de soins : soumettre un fichier, attendre, enregistrer la liste (vous l’utilisez dans la partie 4)
MODEL_CARD.md, LICENSE	Les usages prévus et exclus du modèle; la licence MIT du code

Partie 1 : vérifier le paquet avant de le mettre en service

Charger un fichier joblib peut exécuter du code. Le service vérifie donc d’abord l’empreinte du paquet avec une valeur provenant de l’extérieur du paquet. C’est la méthode de Confiance et intégrité des artefacts. Calculez l’empreinte et comparez-la à celle dans .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

La sortie (macOS) :

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

Dans une vraie équipe, l’empreinte vient du registre de mise en production du modèle, pas d’un fichier à côté du paquet. Ici, cette page tient lieu de registre.

Pour vérifier la provenance, remontez à la source du paquet. training/train.py télécharge les données (3 Mo), vérifie leur empreinte SHA-256 et entraîne de nouveau le modèle. Exécutez-le une fois :

python training/train.py

La sortie :

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

La dernière ligne est identique à celle de model.joblib dans SHA256SUMS. Les mêmes données et le même code donnent le même fichier, octet pour octet. Tout le monde peut vérifier que le paquet contient le modèle décrit dans l’étude de cas d’apprentissage automatique. (Le script réécrit aussi les fichiers de données, avec le même contenu.)

Partie 2 : ce fichier a-t-il besoin d’une tâche?

Le reçu du nettoyeur dans Inférence asynchrone vaut son coût seulement quand l’attente est longue. Mesurez donc d’abord l’attente.

Le chiffre réel

Le projet de départ suit l’idée de Rafael : POST /v1/follow-up-list évalue le fichier pendant que le client attend. Démarrez le service dans le premier terminal :

macOS

export MODEL_SHA256=89056e4cfab4a6a58180461ab01f8b53aeec448ed444ee808699abe2ed6cdf27
fastapi dev

Windows

$env:MODEL_SHA256 = "89056e4cfab4a6a58180461ab01f8b53aeec448ed444ee808699abe2ed6cdf27"
fastapi dev

Linux

export MODEL_SHA256=89056e4cfab4a6a58180461ab01f8b53aeec448ed444ee808699abe2ed6cdf27
fastapi dev

Ouvrez un deuxième terminal (Terminal › New Terminal) et envoyez le fichier de nuit, puis celui de semaine. Sous Windows, tapez curl.exe plutôt que 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='')"

La sortie :

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

Le vrai modèle évalue la semaine entière en environ 0,2 seconde (de 0,19 à 0,22 s dans nos essais). Avec ce modèle, Rafael a raison : si rien ne change, gardez le point de terminaison synchrone simple. Une tâche ajoute du stockage, des interrogations périodiques, de la conservation et davantage de possibilités d’échec.

Ce qui change la réponse

Trois faits à Riverbend changent la réponse :

• L’équipe prévoit un modèle plus lent qui lit aussi chaque lettre de sortie. Nous ne l’avons pas, alors nous simulons son coût. EXTRA_MS_PER_RECORD ajoute ce nombre de millisecondes par ligne (0 en production).
• Une passerelle ferme toute requête après 30 secondes. curl --max-time 30 fait la même chose.
• Personne n’attend à 2 h. Si le service redémarre pendant la nuit, la liste doit encore être là à 7 h 30.

Arrêtez le service (Ctrl+C) et redémarrez-le avec 5 ms supplémentaires par ligne :

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

Le journal indique WARNING readmission: SIMULATED slow model: 5.0 ms more per row (avertissement : modèle lent simulé, 5,0 ms supplémentaires par ligne). Envoyez de nouveau la semaine, avec la limite de la passerelle :

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

La sortie, après 30 secondes :

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

Environ une minute plus tard, le premier terminal affiche :

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

Le client a abandonné après 30 secondes, mais le service ne le savait pas. Il a répondu 200 après 99 secondes, sans destinataire. Le client n’a aucune liste, le service a effectué le travail, et une nouvelle tentative recommencerait tout.

Quand le délai ne suffit plus

Faites une prédiction : combien de millisecondes supplémentaires par ligne font dépasser les 30 secondes au fichier de semaine?

Le temps correspond environ au nombre de lignes multiplié par le temps par ligne : 19 608 × 1,5 ms = 29,4 s. Pour la nuit, même 5 ms par ligne donnent 2 801 × 5 ms = 14 s. Le graphique montre les durées mesurées des tâches :

• Fichier de semaine
• Fichier de nuit
• Limite de 30 s

Durées mesurées des tâches pour le fichier de nuit (2 801 lignes) et celui de semaine (19 608 lignes). À 0, le vrai modèle : 0,04 s et 0,36 s. Au-dessus de 0, un modèle plus lent simulé : chaque milliseconde par ligne ajoute environ 20 s à la semaine.

Description textuelle du graphique

Un graphique linéaire. L’axe horizontal indique le temps supplémentaire par ligne en millisecondes, de 0 à 5. 0 correspond au vrai modèle; 1, 2 et 5 sont simulés. L’axe vertical indique la durée mesurée de la tâche en secondes, de 0 à 100. Le fichier de semaine (19 608 lignes) prend 0,36 s avec le vrai modèle. Avec 1, 2 et 5 ms supplémentaires par ligne, il prend 20,1 s, 40,2 s et 98,8 s. Le fichier de nuit (2 801 lignes) prend 0,04 s, 2,9 s, 5,8 s et 14,1 s. Une ligne horizontale pointillée marque la limite de 30 secondes de la passerelle. La courbe de la semaine la franchit entre 1 et 2 ms par ligne. Celle de la nuit reste sous la limite partout sur le graphique.

Le choix dépend du produit de la taille du fichier et du temps par ligne, pas de l’un ou l’autre seul. Jun crée la tâche pour les fichiers et modèles à venir, et garde la mesure pour justifier son choix. Le coût, comme dans Choisir un mode de service : du stockage, des interrogations périodiques, de la conservation, davantage d’états et environ 270 lignes de code. Il en faut 120 de plus pour survivre à un redémarrage.

Partie 3 : accepter le fichier comme tâche

La vie d’une tâche

Une tâche transforme une longue requête en plusieurs requêtes courtes, comme dans le module 2. Ici, le processus de traitement est un fil d’exécution dans le service. L’appelant interroge régulièrement le service pour connaître l’état.

La vie d'une tâche. La requête fait seulement les vérifications rapides; le processus de traitement calcule les scores. L'appelant interroge le service de nouveau, jamais plus souvent que Retry-After ne le permet.

Description textuelle du diagramme

Lisez de haut en bas. Deux réponses comptent : 202 Accepted (acceptée), qui promet seulement que la tâche existe, et Retry-After (réessayer après). Cette dernière indique à un client respectueux quand interroger le service de nouveau. Chaque étape lente se trouve sur la ligne du processus de traitement. Aucune requête n’attend donc le modèle.

Quatre règles l’accompagnent, chacune représentée par une ligne de code ci-dessous :

• Faites les vérifications rapides avant d’accepter : une colonne manquante produit une erreur maintenant, plutôt qu’une tâche échouée plus tard.
• Limitez la file d’attente : quand 3 tâches attendent, la suivante reçoit 429 Too Many Requests (trop de requêtes).
• Une tâche échouée ne doit pas arrêter le processus de traitement.
• Gardez la liste pendant une durée annoncée : après 48 heures, la tâche indique expired (expirée).

Étape 1 : le module des tâches

Créez un fichier, readmission/jobs.py. Il contient trois parties : l’enregistrement Job, un MemoryStore qui garde les tâches en mémoire, et JobRunner avec son fil de traitement. Voici la première partie :

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

Le stockage est volontairement une classe distincte. Dans la partie 7, vous le remplacez par un stockage qui survit à un redémarrage, sans modifier le lanceur. Ajoutez le lanceur à la fin du fichier :

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 lève QueueFull quand max_queued tâches attendent. run sauvegarde scored après chaque segment de 1 000 lignes, pour que l’état montre la progression. Et except Exception transforme toute erreur en tâche échouée, pour que le processus de traitement passe à la suivante.

Étape 2 : les routes

Ouvrez readmission/routes.py. Remplacez les importations au début par celles-ci :

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

Supprimez ensuite toute la fonction follow_up_list, de @router.post( jusqu’à sa dernière ligne. Ajoutez les routes des tâches à la fin du fichier :

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")

Les vérifications dans submit_job vont de la moins coûteuse à la plus coûteuse : en-tête, taille, colonnes et nombre de lignes, file d’attente. Rien ne lit les valeurs d’une ligne : c’est le travail du processus de traitement.

Étape 3 : démarrer le lanceur

Dans readmission/main.py, ajoutez l’importation sous les autres importations de readmission :

readmission/main.py

from readmission.jobs import JobRunner, MemoryStore

Puis, dans lifespan, après l’appel logger.info(...) qui signale le modèle chargé, démarrez le lanceur :

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,
        )

Le lanceur démarre après que le paquet a franchi ses barrières. Une tâche ne peut donc jamais rencontrer un modèle qui n’est pas prêt.

Étape 4 : essayer

Arrêtez le service. Redémarrez-le sans le coût simulé (sur Windows, faites d’abord Remove-Item Env:EXTRA_MS_PER_RECORD), donc avec fastapi dev. Soumettez ensuite le fichier d’exemple :

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

La sortie :

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}

Demandez la tâche à son adresse location, avec votre propre identifiant de tâche, puis demandez sa liste :

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

La sortie :

{"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

Le 202 est arrivé avant tout calcul de score. Une seconde plus tard, la tâche est done (terminée), nomme son modèle et indique quand sa liste expire. Aucun des 5 séjours n’atteint 0,18.

Remplissez maintenant la file d’attente. Redémarrez le service avec les 5 ms simulées par ligne, comme dans la partie 2. Chaque fichier de nuit prend ainsi 14 secondes. Soumettez ensuite le fichier de nuit cinq fois rapidement, puis une fois de plus avec les en-têtes :

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

Sur Windows PowerShell, la boucle est 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 }.

La sortie (sans les en-têtes date et server) :

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":[]}}

La première tâche démarre immédiatement et les trois suivantes attendent. La cinquième reçoit 429 avec Retry-After: 60, plutôt que de laisser la file d’attente grandir sans limite.

Partie 4 : récupérer la liste et la supprimer à temps

Le client

Le système de l’équipe soignante exécute scripts/follow_up.py. Ouvrez-le. La partie importante est la boucle dans 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")))

Pendant que la tâche attend ou s’exécute, le client attend le nombre de secondes indiqué par Retry-After. Le serveur, et non le client, décide à quelle fréquence il veut être interrogé.

Avec le service toujours à 5 ms par ligne, attendez que les quatre tâches de nuit se terminent (environ une minute). Exécutez ensuite le client :

Second terminal

python scripts/follow_up.py data/discharges_night.csv

La sortie :

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.

Seulement 161 des 2 801 séjours (5,7 %) figurent sur la liste de cette nuit, pas 10 %. Le seuil a été fixé avec d’autres patients. L’équipe d’Adaeze devrait compter les appels chaque nuit.

Comparer la liste aux résultats réels

Redémarrez le service sans le coût simulé et calculez les scores de la semaine. Joignez ensuite la liste à outcomes_week.csv, qui indique qui est revenu dans les 30 jours. Une vraie équipe obtient ces données un mois plus tard.

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.

Utilisez votre propre identifiant de tâche dans le nom du fichier :

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))"

La sortie :

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

La liste donne exactement le résultat du test lors de l’entraînement : 1 866 appels, une précision de 0,270. Cette vérification de bout en bout est plus forte que les 10 cas de parité. Elle couvre tout le parcours, du fichier à la liste.

Supprimer la liste à temps

La condition d’Imani est un paramètre, KEEP_RESULTS_HOURS (48 par défaut). Pour le voir fonctionner, démarrez le service avec 0,001 heure (3,6 secondes) :

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

Soumettez l’exemple, attendez 5 secondes et demandez la liste (utilisez votre identifiant de tâche) :

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

La sortie de la deuxième commande (sans les en-têtes date et server) :

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 (supprimé) n’est pas 404 Not Found (introuvable). Le client distingue « vous avez attendu trop longtemps » de « mauvais identifiant ». Le processus de traitement supprime les listes expirées à son prochain réveil, au plus une minute plus tard. Arrêtez le service (sur Windows, faites aussi Remove-Item Env:KEEP_RESULTS_HOURS).

Partie 5 : un fichier incorrect ne doit pas arrêter le processus de traitement

Les exportations peuvent mal se passer. data/discharges_typos.csv contient 3 cellules incorrectes sur 200 lignes. Dans data/discharges_hours.csv, time_in_hospital est en heures plutôt qu’en jours, dans chaque ligne. Démarrez le service (fastapi dev) et envoyez les deux fichiers, puis l’exemple :

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

La sortie :

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.

Regardez l’état de la première tâche (utilisez votre identifiant de tâche) :

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}

• Quelques lignes incorrectes sont mises de côté, chacune avec son numéro de ligne et son champ. Les 197 autres reçoivent un score.
• Un fichier incorrect échoue au complet : plus de 5 % de lignes incorrectes, aucune liste, comme dans Inférence par lots.
• La tâche suivante s’exécute : la tâche échouée n’a pas arrêté le processus de traitement.

Partie 6 : placer le service dans un conteneur et le redémarrer

Empaquetez le service comme dans Paramètres de ressources du conteneur. Le projet de départ contient requirements-run.txt. Créez trois fichiers. Le 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, pour exclure les données et l’environnement virtuel de l’image :

.dockerignore

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

Et 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

Copiez .env.example dans .env (Windows : Copy-Item .env.example .env). Vérifiez qu’aucun fastapi dev ne tourne sur le port 8000, puis démarrez le conteneur :

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

La sortie de docker compose logs (les dernières lignes) :

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

L’image occupe 787 Mo sur disque (166 Mo compressée), comme le service du cours. Dans le conteneur (1 CPU), la tâche de la semaine a pris de 0,33 à 0,73 seconde avec le vrai modèle.

Simulez maintenant le modèle lent dans le conteneur : ajoutez la ligne EXTRA_MS_PER_RECORD=5 à .env et relancez docker compose up -d --wait. Soumettez la semaine et demandez son état après environ 10 secondes (utilisez votre identifiant de tâche) :

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}

Faites maintenant ce qu’une plateforme fait quand elle déplace un conteneur, applique une mise à jour ou récupère après un plantage : redémarrez-le.

Erreur courante

Après un redémarrage, la tâche a disparu

Symptôme. Vous redémarrez le conteneur, puis demandez la tâche qui était en cours :

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":[]}}

Le journal de l’ancien processus se termine calmement : Application shutdown complete. (arrêt de l’application terminé). Notre nom de conteneur commence par c17cs3-, le nom de notre projet de test. Le vôtre est readmission-service-readmission-1.

Cause. MemoryStore garde les tâches, les fichiers et la progression dans la mémoire d’un seul processus. Un redémarrage termine le processus, et sa mémoire disparaît avec lui.

Solution. Gardez les tâches et les listes hors du processus, dans un stockage qui lui survit. Laissez le nouveau processus retrouver les tâches qui étaient en cours. C’est la partie 7.

Ce que cela signifie pour un vrai service d’IA. Cette défaillance est silencieuse : /ready répond 200, les journaux ne contiennent aucune erreur. L’appelant voit 404 pour une tâche que le service a acceptée avec 202. À 7 h 30, les infirmières n’ont aucune liste, et personne n’a reçu d’alerte. Un 202 est une promesse. Un service qui la tient seulement tant que rien ne redémarre ne la tient pas.

Partie 7 : garder les tâches à l’abri d’un redémarrage

L’idée

Dans un vestiaire, un préposé qui mémorise seulement les crochets perd tout au changement de quart. Un registre au comptoir permet à tout préposé de retrouver n’importe quel manteau. Le registre est un stockage durable : il survit à la personne qui y a écrit.

Pour le service, ce registre est SQLite, une base de données complète dans un seul fichier, incluse dans Python. La base, les fichiers et les listes résident sur un volume Docker, comme dans Données persistantes. Au démarrage, une tâche qui était en cours retourne dans la file d’attente. Ses scores sont recalculés depuis la première ligne.

Le même redémarrage, deux résultats. En mémoire, la tâche disparaît. Dans SQLite sur un volume, elle retourne dans la file d'attente et se termine.

Description textuelle du diagramme

Le redémarrage coûte le temps de la tâche en cours, pas la tâche elle-même. C’est sûr ici pour deux raisons. Le même fichier avec le même paquet donne toujours la même liste. Et le processus de traitement écrit la liste en une seule étape, à la fin.

Étape 1 : le stockage SQLite

Créez un fichier, readmission/store.py. Il a les mêmes méthodes que MemoryStore, donc le lanceur ne change pas :

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)

Deux méthodes font l’essentiel du travail. finish écrit la liste dans un fichier .partial et le renomme. Ainsi, un redémarrage ne laisse jamais une demi-liste : c’est l’écriture atomique d’une tâche par lots. recover s’exécute au démarrage et remplace running (en cours) par queued (en attente).

Étape 2 : choisir le stockage

Dans readmission/settings.py, ajoutez deux champs à la fin de la classe Settings :

readmission/settings.py

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

Dans load_settings, ajoutez deux lignes avant la parenthèse fermante de Settings(...), et une vérification après :

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")

Dans readmission/main.py, ajoutez l’importation from readmission.store import SqliteStore. Remplacez l’appel JobRunner(...) dans lifespan par ceci :

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 reste une option pour les tests, avec un avertissement dans le journal.

Étape 3 : le dossier et le volume

Le conteneur s’exécute sous l’utilisateur app, qui ne peut pas créer de dossiers dans /app. Dans le Dockerfile, ajoutez ces lignes avant 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

Dans compose.yaml, ajoutez les deux paramètres à environment, le volume au service, et le nom du volume à la fin :

compose.yaml

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

Le compose.yaml final du projet terminé montre où placer chaque ligne.

Étape 4 : le prouver

Construisez et démarrez la nouvelle image, avec EXTRA_MS_PER_RECORD=5 toujours dans .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

Sur Windows, utilisez docker compose logs | Select-String "readmission:". Soumettez la semaine, attendez environ 10 secondes, redémarrez et interrogez de nouveau le service (utilisez votre identifiant de tâche) :

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

La sortie des deux requêtes d’état, avant et après le redémarrage :

{"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}

Le journal du nouveau processus indique INFO readmission: jobs are kept in /app/jobs; 1 running job(s) queued again (les tâches sont conservées dans /app/jobs; 1 tâche en cours remise en attente). Récupérez la liste avec le client, à partir de l’identifiant de la tâche :

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.

La tâche a survécu au redémarrage, mais pas sa progression (scored est passé de 2 000 à 0). C’est le prix de cette conception simple.

Un vrai déploiement fait plus qu’un redémarrage : il crée un nouveau conteneur à partir de l’image. Essayez cela aussi :

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

Le nouveau conteneur a trouvé la liste sur le volume. Le fichier d’entrée a été supprimé à la fin de la tâche, comme Imani l’avait demandé.

Les trois emplacements des tâches, mesurés :

Emplacement des tâches	Après un redémarrage	Après un nouveau conteneur
Dans la mémoire du processus (MemoryStore)	Perdues (404)	Perdues (404)
SQLite dans les fichiers du conteneur, sans volume	Conservées (200)	Perdues (404)
SQLite sur un volume	Conservées, une tâche en cours est remise en attente	Conservées

Quand cette conception suffit. Elle suffit pour un seul processus avec un seul fil de traitement. Avec plusieurs processus, chacun démarrerait un fil de traitement et remettrait la tâche en cours de l’autre dans la file d’attente. Plusieurs processus ont besoin d’un service de file d’attente, par exemple Azure Queue Storage ou Amazon SQS. Il confie chaque tâche à un seul processus de traitement à la fois. Les listes vont alors dans un stockage d’objets. Le cours Mise à l’échelle des API et des charges de travail d’IA couvre les deux. Le processus de traitement partage aussi le CPU avec l’API. Pendant la tâche de la semaine, une requête d’état a pris 81 ms plutôt que moins de 1 ms.

Partie 8 : tester

Les tests rendent chaque règle de cette page reproductible. Créez 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"

Dans un test, un « redémarrage » est un deuxième objet d’application avec les mêmes paramètres. Il voit seulement ce que le stockage a conservé. Le dernier test laisse une tâche running, comme le ferait un plantage, et vérifie que le démarrage suivant la termine.

python -m pytest

La sortie (dernières lignes) :

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 ==============================

Les 13 tests des tâches couvrent chaque code visible par l’appelant, chaque état d’une tâche, la règle des 5 % et le redémarrage.

Notre essai complet

Nous avons exécuté chaque étape de cette page depuis starter.zip, dans un nouveau dossier avec un nouvel environnement virtuel. Les commandes et les sorties complètes se trouvent dans le dépôt du cours (reference/c17/case-studies/readmissions-async/RUNS.md).

Vérification	Résultat
Tests	19 réussis au départ; 32 à la fin; projet terminé : 32 réussis
Entraînement	Exactement le même model.joblib que dans le paquet, octet par octet (a7fbea052900…)
Vrai modèle, fichier de la semaine	De 0,19 à 0,22 seconde en une requête; de 0,31 à 0,36 seconde comme tâche; de 0,33 à 0,73 seconde dans le conteneur
Simulation de 5 ms par ligne	La requête a dépassé le délai de 30 secondes; la tâche s’est terminée en 99 secondes
File d’attente, conservation, fichiers incorrects	429 pour la cinquième tâche; 410 après la durée de conservation; 3 lignes rejetées, un fichier incorrect a échoué
Redémarrage	En mémoire : tâche perdue (404). Dans SQLite sur un volume : remise en attente, puis terminée; liste conservée dans un nouveau conteneur
Non testé	Windows et Linux; plusieurs processus; un déploiement dans le nuage

Utilisation responsable

Une liste de suivi semble inoffensive : elle indique seulement qui appeler. Pourtant, les gens agissent selon ces listes, qui peuvent cacher les personnes absentes. Les conditions d’Imani font partie du produit.

La liste ne décide jamais seule. Elle classe les séjours pour proposer un appel. Elle ne décide pas qui peut quitter l’hôpital, qui reçoit moins de soins ou combien une personne paie. Un score faible ne signifie pas « sans risque » : les appels rejoignent seulement 22 % des patients qui reviennent. Environ la moitié des patients qui reviennent n’avaient aucun séjour à l’hôpital dans l’année précédente. Le modèle leur attribue donc généralement un score faible. Le personnel infirmier doit pouvoir ajouter n’importe quel patient à la liste. L’équipe doit l’indiquer dans ses procédures.

Le service prend seulement ce dont il a besoin. Le contrat refuse toute colonne autre que les 19 variables. Le service rejette donc un fichier contenant race (race), gender (genre) ou le payeur avant de créer une tâche (test test_a_sensitive_column_is_refused_before_any_job). Les fichiers ne contiennent ni noms ni numéros de patient. Le service supprime le fichier d’entrée quand sa tâche se termine, et la liste après 48 heures. Le journal contient des identifiants de tâche et des nombres, jamais une ligne de données.

Les données sont anciennes et viennent d’un seul pays. Le modèle a appris à partir de données d’hôpitaux américains datant de 1999 à 2008, avec des codes ICD-9. Les soins, le codage et les assurances ont changé depuis. Avant toute utilisation réelle, un hôpital doit l’entraîner et le tester avec des données récentes de ses propres patients. L’étude de cas en apprentissage automatique a aussi montré qu’il fonctionne moins bien pour les patients les plus âgés. Chaque mois, vérifiez la précision et le rappel par âge, race, genre et payeur, avec les résultats dès leur arrivée.

Le signal est faible. Une aire sous la courbe ROC de 0,663 est loin de 1. La liste fait mieux que le hasard : une précision de 0,27 contre 0,11. Elle fait seulement un peu mieux qu’une règle simple : appeler les patients ayant le plus de séjours dans l’année précédente. Pour les mêmes patients, l’étude de cas en apprentissage automatique a mesuré 0,264 pour les premiers 10 % de la liste du modèle. Elle a mesuré 0,247 pour la règle. Ce n’est pas une prédiction pour un patient individuel.

Ce qu’il faudrait aussi pour un vrai service. Des lois comme HIPAA aux États-Unis et le RGPD en Europe encadrent les données de santé. Ceci n’est pas un avis juridique. Un vrai déploiement exige un contrôle d’accès pour chaque utilisateur et un journal d’audit indiquant qui a récupéré quelle liste. Il exige aussi le chiffrement du volume et une révision par l’équipe de gouvernance de l’information de l’hôpital. Cette étude de cas ne met rien de cela en place; elle développe les habitudes qui le rendent possible.

Le coût simulé n’est pas un modèle. EXTRA_MS_PER_RECORD montre quand ce mode de service atteint ses limites. Il ne montre pas le comportement d’un vrai modèle de texte : mesurez ce modèle avant de choisir ses limites.

Faites-en un projet de portfolio

Ce projet montre que vous pouvez offrir un modèle dont le travail ne tient pas dans une requête. Il montre aussi que vous testez ce qui arrive lors des redémarrages. Publiez-le dans un dépôt, sous votre nom.

Quoi mettre dans le dépôt :

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. Partez de votre dossier de projet ou du projet terminé. Vérifiez que .gitignore contient .venv/, .env, jobs/, out/ et data/raw/.
2. Créez le dépôt : git init, puis git add . et git commit -m "Readmission scoring service with queued jobs".
3. Avec git ls-files, vérifiez que le paquet et data/README.md y sont, mais qu’aucun fichier .env n’y est.
4. Envoyez-le sur GitHub comme dépôt public. Gardez LICENSE (MIT) pour le code; les données conservent leur propre licence (CC BY 4.0). Le README l’indique avec l’attribution.
5. Réécrivez le premier paragraphe du README dans vos mots. Ajoutez vos propres mesures : les durées des tâches et le test de redémarrage.

Un modèle de README :

# 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.

Comment le présenter en entrevue : « J’ai rendu un modèle de réadmission accessible par une API de tâches asynchrones. Le programme appelant reçoit le code 202 avec une adresse de tâche. Il vérifie l’état avec Retry-After et récupère la liste. Le service a une limite de file d’attente, des règles d’échec partiel et une durée de conservation. Les tâches en mémoire disparaissaient au redémarrage du conteneur. Je les ai donc déplacées vers SQLite sur un volume. Un test prouve qu’une tâche survit à un redémarrage. »

Pour aller plus loin :

1. Une clé d’idempotence. Si le programme d’exportation de Rafael perd la réponse 202 et renvoie le fichier, le service crée deux tâches. Acceptez un en-tête Idempotency-Key et retournez la tâche existante pour une clé déjà reçue.
2. Un rappel au lieu de vérifications répétées. Laissez le client fournir une URL, puis appelez-la quand la tâche se termine. Gardez les vérifications répétées comme solution de repli.
3. Un processus de travail distinct. Déplacez le processus de travail dans son propre conteneur, qui lit les tâches dans le même stockage. Mesurez la latence des requêtes d’état pendant l’exécution d’une grosse tâche.
4. Des raisons pour chaque appel. La régression logistique peut décomposer chaque score selon la contribution des variables. L’étude de cas en apprentissage automatique montre comment. Ajoutez les deux principales contributions à chaque ligne de la liste, pour que le personnel infirmier puisse amorcer l’appel avec elles.
5. Un rapport mensuel par groupe. Joignez les listes aux résultats et présentez la précision et le rappel par tranche d’âge. Suivez l’exemple de l’étude de cas en apprentissage automatique.

Vérification des connaissances

Vérifiez votre compréhension

Rafael dit : le vrai modèle traite la semaine en 0,2 seconde, donc la tâche est inutile. Quand a-t-il raison, et pourquoi garderiez-vous quand même la tâche?

Afficher la réponse

Il a raison pour le modèle et les fichiers actuels. Une requête synchrone de 0,2 seconde est plus simple. Une tâche ajoute du stockage, des vérifications répétées, une durée de conservation et davantage d’états d’échec. Gardez la tâche si le travail multiplié par la taille peut dépasser le temps alloué par le programme appelant. Le fichier hebdomadaire mesuré dépasse la limite de 30 secondes de la passerelle avec environ 1,5 ms supplémentaires par ligne. L’équipe prévoit aussi un modèle plus lent. Gardez-la également parce que le programme appelant fonctionne sans surveillance à 2 h. Avec une tâche, la liste survit à un redémarrage et attend jusqu’à 7 h 30. Une réponse synchrone disparaît avec la connexion. La décision repose sur les mesures : consignez-les et mesurez de nouveau quand le modèle change.

Vérifiez votre compréhension

Au démarrage, recover() remet une tâche en cours dans la file d'attente. Pourquoi peut-on recalculer les scores depuis la première ligne sans risque, et qu'est-ce qui rendrait cela risqué?

Afficher la réponse

C’est sans risque parce que la tâche est répétable. Le stockage conserve le fichier d’entrée jusqu’à la fin de la tâche, et l’empreinte fixe le paquet utilisé. Les mêmes lignes avec le même paquet donnent toujours les mêmes scores. Le processus de travail écrit aussi la liste en une seule étape : un fichier .partial, puis un renommage. Une exécution inachevée ne laisse donc pas de liste partielle. Ce serait risqué si la tâche avait des effets qui ne doivent pas se produire deux fois. Par exemple, un message à chaque patient. Ce serait aussi risqué si le modèle ou l’entrée pouvait changer entre les deux exécutions. La tâche devrait alors consigner ce qu’elle a déjà fait.

Vérifiez votre compréhension

Rafael veut plus de débit et lance le service avec deux processus, fastapi run --workers 2, et le stockage SQLite. Prévoyez deux problèmes.

Afficher la réponse

Chaque processus démarre son propre fil d’exécution de travail. D’abord, chaque processus exécute recover() au démarrage. Un processus qui redémarre remet donc la tâche en cours de l’autre processus dans la file d’attente. La tâche s’exécute alors deux fois, en même temps. Ensuite, les deux fils peuvent lire la même tâche, la plus ancienne en attente, avant que l’un écrive l’état running. Deux fils de travail peuvent donc prendre la même tâche. La limite de file d’attente reste partagée, car les deux processus comptent les entrées de la même table. Pour cette conception, la solution est un seul processus avec un seul fil de travail. Pour plusieurs processus, utilisez un service de file d’attente. Il attribue chaque tâche à un seul processus de travail, puis à un autre seulement si le premier ne répond plus.

Récapitulatif

• Une tâche vaut son coût seulement si le travail multiplié par la taille peut dépasser le temps alloué par le programme appelant. Mesurez d’abord. Le vrai modèle a calculé les scores de 19 608 séjours en environ 0,2 seconde. Un coût simulé de 5 ms par ligne a pris 99 secondes, dépassant la limite de 30 secondes de la passerelle.
• La requête fait seulement des vérifications rapides (type, taille, colonnes, lignes, file d’attente). Elle répond 202 Accepted (acceptée) avec Location (adresse) et Retry-After (délai avant une nouvelle tentative). Le processus de travail calcule les scores par segments et indique la progression.
• Une file d’attente pleine répond 429 avec Retry-After. Quelques lignes incorrectes sont mises de côté par numéro de ligne; au-delà de 5 %, la tâche échoue. Une tâche échouée n’arrête jamais le processus de travail.
• Une liste est conservée pendant une durée annoncée, puis supprimée : 410 Gone (ressource supprimée) indique pourquoi au client.
• Les tâches en mémoire disparaissent au redémarrage, sans avertissement. Les tâches dans SQLite sur un volume survivent à un redémarrage et à un nouveau conteneur. Une tâche en cours retourne dans la file d’attente.
• Un processus, un fil de travail : plusieurs processus exigent un service de file d’attente et du stockage partagé.
• La liste aide l’équipe soignante à faire ses appels. Elle ne décide jamais seule pour un patient et prend seulement les colonnes nécessaires. Ses données sont anciennes : réentraînez le modèle et vérifiez-le d’abord avec vos propres patients.

Nettoyez. Arrêtez le conteneur et supprimez son volume, ce qui supprime aussi les tâches et les listes : docker compose down -v. Supprimez l’image : docker image rm readmission-service:1.0.0.

Leçon suivante : Travail final