@atomic-ehr/codegen : profils US Core en Python

Construire des ressources US Core à la main est fastidieux. On estampille meta.profile, on cherche les codes LOINC, on code à la main l'extension imbriquée us-core-race — chaque champ est une faute de frappe en attente, chaque profil est sa propre version du même cérémonial.

@atomic-ehr/codegen fait disparaître ce code générique. Pointez-le vers le US Core IG et vous obtenez un modèle Pydantic par type de base, plus une classe enveloppe Python ordinaire par profil, avec des accesseurs typés pour les valeurs fixes, les extensions et les tranches, ainsi qu'une méthode validate() qui sait ce que le profil exige.

Ce tutoriel parcourt le processus de bout en bout sur deux profils US Core : US Core Patient et US Core Blood Pressure.

Ce que vous allez construire

Un convertisseur CSV vers FHIR, construit étape par étape :

1. générer des classes de profils pour US Core Patient et US Core Blood Pressure à partir de hl7.fhir.us.core@8.0.1,
2. transformer chaque ligne en un US Core Patient — accesseurs d'extension typés et apply(),
3. transformer chaque ligne en un US Core Blood Pressure — tranches typées, LOINC fixe et validate(),
4. les empaqueter dans un Bundle,
5. relire le bundle avec des accesseurs typés pour calculer une pression artérielle moyenne,
6. publier le bundle sur un serveur Aidbox local via le client fhirpy.

Prérequis

• Node.js 20+ (ou Bun) — le générateur lui-même est le paquet Node @atomic-ehr/codegen. Vous l'exécutez une fois pour produire le Python; après cela, vous n'avez plus besoin de Node. Le script de génération est quelques lignes de TypeScript (présentées ci-dessous).
• Python 3.12+ — le code généré cible le Python moderne (PEP 604 unions X | None, modèles génériques via typing_extensions).
• Pydantic v2 (pydantic>=2.11) et fhirpy — les modèles générés sont en Pydantic v2; avec le client fhirpy par défaut, ils s'intègrent également au client asynchrone de fhirpy (étape 6). Les deux versions sont épinglées dans le fichier requirements.txt généré (voir étape 1); passez client: "none" pour du Pydantic pur sans code client.
• Une connaissance de base de FHIR et de US Core (savoir ce que signifient « profil » et « tranche » suffit).

Étape 1 — Générer les classes de profils

La génération de code s'exécute via l'outil Node; configurez donc un petit projet générateur à côté de votre application Python :

mkdir py-us-core-tutorial && cd py-us-core-tutorial
npm init -y
npm install --save-dev @atomic-ehr/codegen tsx typescript

Créez generate.ts :

import { APIBuilder, mkCodegenLogger, prettyReport } from "@atomic-ehr/codegen";

const main = async () => {
  const logger = mkCodegenLogger({
    suppressTags: ["#fieldTypeNotFound", "#duplicateSchema", "#duplicateCanonical", "#largeValueSet"],
  });

  const builder = new APIBuilder({ logger })
    .fromPackage("hl7.fhir.us.core", "8.0.1")
    .typeSchema({
      treeShake: {
        "hl7.fhir.us.core": {
          "http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient": {},
          "http://hl7.org/fhir/us/core/StructureDefinition/us-core-blood-pressure": {},
        },
        "hl7.fhir.r4.core": {
          "http://hl7.org/fhir/StructureDefinition/Bundle": {},
        },
      },
    })
    .python({
      generateProfile: true,
      allowExtraFields: false,
      primitiveTypeExtension: true,
    })
    .outputTo("./fhir_types")
    .cleanOutput(true);

  const report = await builder.generate();
  console.log(prettyReport(report));
  if (!report.success) process.exit(1);
};

main();

Les paramètres importants ici :

• generateProfile: true — émet une classe enveloppe par profil avec des accesseurs typés pour les extensions, les tranches et les valeurs fixes. Sans cette option, vous n'obtenez que les modèles Pydantic R4 de base.
• allowExtraFields: false — les modèles générés utilisent extra="forbid" de Pydantic, de sorte qu'un champ inconnu déclenche une erreur à l'analyse plutôt que d'être silencieusement ignoré.
• primitiveTypeExtension: true — génère également les compléments d'extension primitive FHIR (les compagnons _field, p. ex. birthDateExtension) afin que vous puissiez attacher des extensions et des id aux valeurs primitives.
• treeShake: { ... } — seuls les canoniques listés et leurs dépendances transitives sont générés (~20 fichiers au lieu de centaines).

Exécutez-le. prettyReport(report) imprime un résumé groupé afin que vous voyiez ce qui a été émis sans avoir à parcourir le répertoire de sortie :

$ npx tsx generate.ts
# journaux de génération omis; voici le résumé prettyReport
Generated files (24 files, 12 kloc):
  python (23 files, 3.6 kloc):
    - fhir_types/ (4 files, 622 loc)
    - fhir_types/hl7_fhir_r4_core/ (8 files, 1.1 kloc)
    - fhir_types/hl7_fhir_r4_core/profiles/ (2 files, 164 loc)
    - fhir_types/hl7_fhir_us_core/profiles/ (9 files, 1.8 kloc)
  ir-report (1 files, 8.2 kloc):
    - fhir_types/README.md (8223 loc)
Duration: 8097ms
Status: 🟩 Success

La structure sur disque ressemble à ceci :

fhir_types/
├── hl7_fhir_r4_core/                  # Base R4 Pydantic models
│   ├── base.py                        # Element, Coding, CodeableConcept, Quantity, ...
│   ├── resource.py                    # Resource, DomainResource, Meta
│   ├── patient.py
│   ├── observation.py
│   ├── bundle.py
│   ├── profiles/                      # base R4 profiles US Core builds on
│   │   └── observation_observation_vitalsigns.py   # vital-signs base (BP derives from it)
│   └── ...
├── hl7_fhir_us_core/
│   └── profiles/
│       ├── __init__.py                # re-exports the profile classes
│       ├── patient_uscore_patient_profile.py
│       ├── observation_uscore_blood_pressure_profile.py
│       ├── extension_uscore_race_extension.py
│       └── ...
├── fhirpy_base_model.py               # fhirpy client base model (default fhirpy client)
├── profile_helpers.py                 # Runtime helpers shared by all profile classes
├── README.md                          # IR report — human-readable dump of the generated types
└── requirements.txt                   # pydantic, fhirpy (+ pytest, requests for tests/Step 6)

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

Pointez votre application Python vers le répertoire fhir_types/ émis et installez les dépendances :

pip install -r fhir_types/requirements.txt

Le fichier requirements.txt généré épingle Pydantic et fhirpy, ainsi que pytest et requests pour les tests et les exemples.

Le code complet du tutoriel se trouve dans Aidbox/examples — generate.ts, load.py, avg.py, post.py, le CSV et le répertoire fhir_types/ versionné afin que vous puissiez parcourir le code généré sans exécuter le générateur. Pour une exploration plus large de l'API des profils, le dépôt codegen contient également un exemple de test python-r4-us-core. Les deux utilisent les noms d'attributs camelCase par défaut, exactement comme les extraits ici. (Passez fieldFormat: "snake_case" si vous préférez écrire les attributs birth_date, effective_date_time; la sérialisation émet toujours du JSON FHIR en camelCase correct dans les deux cas.)

Étape 2 — Ligne vers un US Core Patient

Le fichier d'entrée est patients.csv — données démographiques de base plus une lecture de pression artérielle par patient. La race utilise les codes de catégorie OMB attendus par US Core :

mrn,family,given,birthDate,gender,raceCode,raceDisplay,effectiveDateTime,systolic,diastolic
MRN-001,Lovelace,Ada,1815-12-10,female,2106-3,White,2026-04-15,120,80
MRN-002,Turing,Alan,1912-06-23,male,2106-3,White,2026-04-15,118,76
MRN-003,Curie,Marie,1867-11-07,female,2106-3,White,2026-04-16,125,82
MRN-004,Carver,George,1864-01-01,male,2054-5,Black or African American,2026-04-16,135,88
MRN-005,Ochoa,Ellen,1958-05-10,female,2054-5,Black or African American,2026-04-17,128,84

csv.DictReader transmet chaque ligne sous forme de dict[str, str] ordinaire; l'analyse des valeurs numériques se produit plus tard, lorsqu'on passe les valeurs aux accesseurs de profil typés.

Le profil US Core Patient ajoute quelques extensions et rend identifier et name obligatoires. La classe générée possède un accesseur typé pour chacun :

from fhir_types.hl7_fhir_r4_core.base import Identifier, HumanName, Coding
from fhir_types.hl7_fhir_r4_core.patient import Patient
from fhir_types.hl7_fhir_us_core.profiles import UscorePatientProfile

def row_to_patient(row: dict[str, str]) -> UscorePatientProfile:
    base_patient = Patient(
        resourceType="Patient",
        identifier=[Identifier(system="http://hospital.example.org/mrn", value=row["mrn"])],
        name=[HumanName(family=row["family"], given=[row["given"]])],
        gender=row["gender"],          # gender is a Literal type — Pydantic validates the value
        birthDate=row["birthDate"],    # default camelCase attrs match the FHIR wire names
    )

    patient = UscorePatientProfile.apply(base_patient)

    patient.set_race({
        "ombCategory": {"system": "urn:oid:2.16.840.1.113883.6.238", "code": row["raceCode"], "display": row["raceDisplay"]},
        "text": row["raceDisplay"],
    })

    return patient

Deux phases :

1. Construire le Patient ordinaire — les champs obligatoires par le profil (identifier, name) et les champs à support obligatoire (gender, birthDate) comme un modèle Pydantic R4 typé. Construire avec les noms d'attributs camelCase par défaut (birthDate, les noms de transmission FHIR); les valeurs sont validées immédiatement (p. ex. gender est un Literal["male", "female", "other", "unknown"]).
2. Puis UscorePatientProfile.apply(base_patient) estampille meta.profile et retourne une instance de profil avec des accesseurs typés pour les extensions US Core. apply() enveloppe la ressource sur place — le profil mute le même objet Patient.

Trois remarques sur ce que l'API de profil fait pour vous :

• Trois formes d'accesseurs d'extension. set_race({ "ombCategory": ..., "text": ... }) accepte une entrée plate — notez que les clés de sous-extension (ombCategory, detailed, text) sont les noms de tranche en camelCase — et génère la plomberie extension[] imbriquée. Le même accesseur accepte également une instance de profil d'extension typée (UscoreRaceExtension) ou une Extension brute, et lève une erreur si l'url d'une extension brute ne correspond pas.
• Les extensions à valeur unique prennent directement la valeur. us-core-individual-sex porte un seul valueCoding, donc set_sex(Coding(code="female")) prend un Coding (ou une Extension brute).
• Pas d'accesseurs pour les champs de base à support obligatoire. gender, birthDate et address ne sont pas profilés davantage par US Core, donc la classe de profil n'émet pas d'enveloppes de style .set_gender() — remplissez-les comme des champs Patient ordinaires. validate() avertit quand même si un champ à support obligatoire est manquant.

Pydantic émet un UserWarning lorsqu'une liste extension[] contient des dicts ordinaires plutôt que des instances Extension — prévu avec la plomberie de dict plat actuelle. Réduisez-le au silence avec warnings.filterwarnings("ignore", category=UserWarning, module="pydantic").

Étape 3 — Ligne vers un US Core Blood Pressure

Le profil BP est là où codegen rapporte vraiment sa valeur. Le profil US Core Blood Pressure :

• fixe code au LOINC 85354-9 (« Blood pressure panel »),
• fixe une tranche de catégorie vital-signs,
• définit des tranches component[systolic] et component[diastolic] avec des discriminateurs LOINC spécifiques (8480-6 et 8462-4),
• exige un effectiveDateTime ou un effectivePeriod,
• exige un valueQuantity à l'intérieur de chaque tranche.

Coder cela à la main pour chaque ligne est exactement le genre de chose que codegen élimine. La classe générée le réduit à trois accesseurs :

from fhir_types.hl7_fhir_r4_core.base import Reference
from fhir_types.hl7_fhir_us_core.profiles import UscoreBloodPressureProfile

def row_to_bp(row: dict[str, str], patient_urn: str) -> UscoreBloodPressureProfile:
    bp = UscoreBloodPressureProfile.create(
        status="final",
        subject=Reference(reference=patient_urn),
    )

    (
        bp.set_effective_date_time(row["effectiveDateTime"])
          .set_systolic({"value": float(row["systolic"]), "unit": "mmHg", "system": "http://unitsofmeasure.org", "code": "mm[Hg]"})
          .set_diastolic({"value": float(row["diastolic"]), "unit": "mmHg", "system": "http://unitsofmeasure.org", "code": "mm[Hg]"})
    )

    errors = bp.validate()["errors"]
    if errors:
        raise ValueError(f"{row['mrn']}: {'; '.join(errors)}")

    return bp

Ce qui se passe en coulisse :

• create() fait le cérémonial. Il estampille meta.profile, remplit le code fixe (LOINC 85354-9), ajoute la tranche de catégorie vital-signs et ajoute des ébauches vides component[systolic] / component[diastolic] avec les codes discriminateurs déjà définis. create() prend des arguments à mot-clé uniquement; create_resource() est identique mais retourne une Observation ordinaire plutôt qu'un enveloppe de profil.
• set_systolic({ "value": ..., "unit": ... }) remplit le valueQuantity à l'intérieur de la tranche systolique. Le code discriminateur sur ce composant est déjà là grâce à create() — vous ne fournissez que la lecture.
• validate() retourne {"errors": [...], "warnings": [...]}. Les erreurs bloquent (champs obligatoires, champs exclus, variantes de choix non autorisées, cardinalité des tranches). Les avertissements signalent des préoccupations de support obligatoire. Une ligne malformée échoue rapidement avec le numéro de dossier — vous ne le découvrez pas au moment du POST.

Vous n'avez pas saisi les codes discriminateurs. Vous n'avez pas mémorisé 85354-9. Les accesseurs s'enchaînent de façon fluide (chacun retourne le profil), exactement comme l'API TypeScript.

Étape 4 — Assembler le Bundle

Chaque ligne produit un Patient et une Observation BP liés par l'espace réservé urn:uuid du Patient. Empaquetez-les sous forme d'entrées de transaction. Les classes générées Bundle et BundleEntry sont génériques par rapport à la ressource contenue, de sorte qu'un Bundle[Patient | Observation] garde entry[].resource typé à cette union. (row_to_patient et row_to_bp sont les fonctions des étapes 2-3; dans l'exemple, elles se trouvent toutes dans un fichier load.py, elles sont donc déjà dans la portée ici.)

import json
import csv
import uuid

from fhir_types.hl7_fhir_r4_core.patient import Patient
from fhir_types.hl7_fhir_r4_core.observation import Observation
from fhir_types.hl7_fhir_r4_core.bundle import Bundle, BundleEntry, BundleEntryRequest

def row_to_entries(row: dict[str, str]) -> list[BundleEntry[Patient | Observation]]:
    patient_urn = f"urn:uuid:{uuid.uuid4()}"
    patient = row_to_patient(row)
    bp = row_to_bp(row, patient_urn)

    return [
        BundleEntry(fullUrl=patient_urn, resource=patient.to_resource(),
                    request=BundleEntryRequest(method="POST", url="Patient")),
        BundleEntry(fullUrl=f"urn:uuid:{uuid.uuid4()}", resource=bp.to_resource(),
                    request=BundleEntryRequest(method="POST", url="Observation")),
    ]

rows = list(csv.DictReader(open("patients.csv")))
print(f"Loaded {len(rows)} rows")

entries = [entry for row in rows for entry in row_to_entries(row)]

bundle = Bundle[Patient | Observation](
    resourceType="Bundle",
    type="transaction",
    entry=entries,
)

with open("bundle.json", "w") as f:
    json.dump(bundle.model_dump(by_alias=True, exclude_none=True), f, indent=2)
print(f"Wrote bundle with {len(entries)} entries")

$ python load.py
Loaded 5 rows
Wrote bundle with 10 entries

À noter :

• to_resource() vous donne le modèle ordinaire — la ressource Pydantic sous-jacente, sans enveloppe, prête à être insérée dans un BundleEntry.
• model_dump(by_alias=True, exclude_none=True) produit du JSON FHIR — by_alias sérialise via les alias de transmission FHIR (de sorte qu'une construction snake_case émet quand même effectiveDateTime) et exclude_none supprime les champs à valeur None. L'appel de sérialisation unique que vous utiliserez partout.
• Références urn:uuid. Le fullUrl du patient et le subject.reference de l'observation partagent un UUID; le serveur le résout en un vrai identifiant lors de la validation.

Étape 5 — Relire : pression artérielle moyenne à partir du Bundle

Relisons maintenant. Analysez bundle.json et calculez les moyennes systolique/diastolique pour exercer l'API côté lecture :

import json
from typing import Any

from fhir_types.hl7_fhir_r4_core.observation import Observation
from fhir_types.hl7_fhir_us_core.profiles import UscoreBloodPressureProfile

bundle = json.load(open("bundle.json"))

def is_us_core_bp(resource: dict[str, Any]) -> bool:
    return (
        resource.get("resourceType") == "Observation"
        and UscoreBloodPressureProfile.canonical_url in (resource.get("meta", {}).get("profile") or [])
    )

bps = [
    UscoreBloodPressureProfile.from_resource(Observation.model_validate(entry["resource"]))
    for entry in bundle.get("entry", [])
    if is_us_core_bp(entry["resource"])
]

def avg(xs: list[float]) -> float:
    return sum(xs) / len(xs)

# get_systolic()/get_diastolic() are Optional, so guard with a walrus before indexing.
systolic = [s["value"] for bp in bps if (s := bp.get_systolic()) is not None]
diastolic = [d["value"] for bp in bps if (d := bp.get_diastolic()) is not None]

print(f"Avg BP: {avg(systolic):.1f}/{avg(diastolic):.1f} mmHg (n={len(bps)})")

$ python avg.py
Avg BP: 125.2/82.0 mmHg (n=5)

Trois choses que le profil fait ici :

• from_resource(obs) valide lors de l'enveloppement. Il vérifie que meta.profile inclut l'URL canonique et retourne une instance de profil, levant une erreur si une ressource qui prétend être le profil est malformée — ainsi un bundle brisé échoue à la lecture, pas au prochain accès de champ.
• Pas de garde-type intégré. Contrairement au prédicat is() de l'API TypeScript, les classes Python ne comportent pas de garde de style .filter(). Vous sélectionnez les candidats vous-même — vérifiez resourceType et canonical_url in meta.profile (ci-dessus), ou enveloppez from_resource() dans un bloc try/except ValueError. Dans les deux cas, canonical_url est exposé comme attribut de classe exactement pour cela.
• get_systolic() / get_diastolic() retournent la valeur plate de la tranche. Pas besoin de parcourir component[].code.coding[].code pour faire correspondre les codes LOINC — le profil sait déjà quelle tranche est laquelle et vous remet les données Quantity sous forme de dict ordinaire.

Voilà le cycle complet : CSV → profils typés → Bundle validé → relecture typée avec des accesseurs tenant compte des profils. Les mêmes quelques lignes traiteraient des pression artérielles récupérées depuis un serveur FHIR, chargées depuis un fichier ou reçues via un abonnement — le profil typé est la forme commune, quelle que soit la source.

Étape 6 — Déposer votre Bundle sur un serveur FHIR

Le pipeline typé n'est que la moitié de l'histoire. Pour voir réellement la transaction validée — identifiants de patient assignés, références urn:uuid réécrites, ressources stockées et interrogeables — vous avez besoin d'un serveur FHIR. Démarrez et exécutez Aidbox :

curl -JO https://aidbox.app/runme && docker compose up -d

Ouvrez http://localhost:8080 dans votre navigateur pour obtenir une licence développeur gratuite, puis extrayez le secret du client racine de docker-compose.yaml dans une variable d'environnement — réutilisée par les commandes curl et le script Python ci-dessous :

export BOX_ROOT_CLIENT_SECRET=$(awk '/BOX_ROOT_CLIENT_SECRET:/{print $2}' docker-compose.yaml)

Vérifiez que le point de terminaison FHIR est actif :

curl -u "root:$BOX_ROOT_CLIENT_SECRET" http://localhost:8080/fhir/metadata

Vous devriez voir un CapabilityStatement JSON.

Envoyez le fichier bundle.json que vous venez d'écrire avec le client asynchrone de fhirpy :

import asyncio
import base64
import json
import os

from fhirpy import AsyncFHIRClient
from fhir_types.hl7_fhir_r4_core import Bundle

secret = os.environ["BOX_ROOT_CLIENT_SECRET"]  # exported above
auth = base64.b64encode(f"root:{secret}".encode()).decode()


async def main() -> None:
    client = AsyncFHIRClient("http://localhost:8080/fhir", authorization=f"Basic {auth}")
    bundle = json.load(open("bundle.json"))
    resp: Bundle = await client.execute("/", method="post", data=bundle)
    if resp.entry is None:
        return

    for entry in resp.entry:
        if entry.response is None:
            continue
        print(entry.response.status, entry.response.location)


asyncio.run(main())

Aidbox retourne un bundle transaction-response — une entrée par entrée d'entrée, chacune avec un 201 Created et un location pointant vers la ressource stockée :

$ python post.py
201 Created Patient/<id>/_history/1
201 Created Observation/<id>/_history/1
...

Interrogez une observation en retour et examinez son subject :

$ curl -u "root:$BOX_ROOT_CLIENT_SECRET" \
  "http://localhost:8080/fhir/Observation?code=http://loinc.org|85354-9" \
  | jq '.entry[].resource.subject.reference'
"Patient/01J..."
"Patient/01J..."

Plus de urn:uuid — Aidbox a réécrit les espaces réservés de façon atomique lors de la validation.

Vérification de types du pipeline

Les modèles générés sont en Pydantic v2, donc le convertisseur se vérifie avec mypy — déjà épinglé dans le fichier requirements.txt généré. Une exigence : activez le plugin mypy de Pydantic (il est fourni avec Pydantic, pas d'installation supplémentaire), sinon mypy ne peut pas déterminer qu'un défaut Field(None, ...) rend un champ optionnel et vous inonde de faux avertissements « argument manquant » sur chaque modèle que vous construisez.

Déposez un fichier mypy.ini à côté de votre code :

[mypy]
strict = True
plugins = pydantic.mypy

Puis exécutez-le :

$ mypy .
Success: no issues found in 35 source files

Vos deux modules de conversion — load.py, avg.py, post.py — et le paquet fhir_types/ généré reviennent tous propres. Les usines typées, to_resource() et le générique Bundle[Patient | Observation] passent tous la vérification, de sorte qu'un mauvais type de champ ou un argument obligatoire manquant est détecté avant d'atteindre le serveur. La couche de profil générée se vérifie aussi sous --strict complet — tout l'intérêt du mypy.ini tenant en ces deux lignes : pas de disable_error_code, pas de strict_optional = False, rien de modifié à la main dans fhir_types/. La seule chose que votre propre code fournit est une garde None ordinaire lors de la lecture d'un champ optionnel — le morse dans avg.py ci-dessus, ou if entries is None: ... avant d'indexer une liste. C'est du Python strict ordinaire, pas un défaut du générateur.

Pour aller plus loin

• Plus d'API de profil. D'autres usines, accesseurs et formes de tranches/extensions sont exercés dans les tests d'exemples codegen : test_profile_patient.py, test_profile_bp.py, test_profile_bodyweight.py et test_profile_typed_bundle.py.
• Bundles typés avec des tranches d'entrée nommées. Un Bundle profilé génère des accesseurs par tranche (set_patient_entry, get_organization_entry) avec une cardinalité unique ou illimitée (max: *) gérée pour vous — voir le test de bundle typé ci-dessus.
• Ajuster la sortie à votre base de code. fieldFormat (snake_case/camelCase), client ("fhirpy"/"none"), allowExtraFields et primitiveTypeExtension sont tous des paramètres de .python({ ... }).
• Mélanger des profils de plusieurs paquets. APIBuilder.fromPackage() s'enchaîne — US Core aux côtés de votre IG personnalisé ou d'une base régionale. localStructureDefinitions() charge des profils directement depuis un dossier de fichiers JSON StructureDefinition.

Conclusion

Le générateur émet à la fois les modèles Pydantic R4 de base et une mince couche de classes de profil par-dessus — pas de DSL à l'exécution, pas d'ORM, pas de cadriciel. to_resource() vous donne toujours une ressource Pydantic ordinaire, et model_dump(by_alias=True, exclude_none=True) vous donne toujours du JSON FHIR ordinaire que vous pouvez envoyer à n'importe quel serveur.

@atomic-ehr/codegen est sous licence MIT; les signalements de bogues et les demandes de tirage sont les bienvenus.

GitHub | NPM | US Core IG