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

Construir recursos US Core a mano es tedioso. Se estampa meta.profile, se buscan los códigos LOINC, se enrolla a mano la extensión anidada us-core-race — cada campo es un error tipográfico esperando ocurrir, cada perfil es su propia versión de la misma ceremonia.

@atomic-ehr/codegen hace desaparecer ese código repetitivo. Apúntelo al US Core IG y obtendrá un modelo Pydantic por tipo base más una clase envolvente Python pura por perfil, con accesores tipados para valores fijos, extensiones y slices, y un método validate() que conoce los requisitos del perfil.

Este tutorial recorre ese proceso de extremo a extremo con dos perfiles US Core: US Core Patient y US Core Blood Pressure.

Qué Va a Construir

Un conversor de CSV a FHIR, construido paso a paso:

1. generar clases de perfil para US Core Patient y US Core Blood Pressure a partir de hl7.fhir.us.core@8.0.1,
2. convertir cada fila en un US Core Patient — setters de extensión tipados y apply(),
3. convertir cada fila en un US Core Blood Pressure — slices tipados, LOINC fijo y validate(),
4. empaquetarlos como un Bundle,
5. leer el bundle de vuelta con getters tipados para calcular una media de TA,
6. publicar el bundle en un servidor Aidbox local mediante el cliente fhirpy.

Requisitos Previos

• Node.js 20+ (o Bun) — el generador en sí es el paquete Node @atomic-ehr/codegen. Se ejecuta una vez para emitir Python; después no se necesita Node. El script de generación son unas pocas líneas de TypeScript (se muestran a continuación).
• Python 3.12+ — el código generado está orientado a Python moderno (PEP 604 uniones X | None, modelos genéricos vía typing_extensions).
• Pydantic v2 (pydantic>=2.11) y fhirpy — los modelos generados son Pydantic v2; con el cliente fhirpy predeterminado también se integran en el cliente asíncrono de fhirpy (Paso 6). Ambos están fijados en el requirements.txt generado (véase el Paso 1); pase client: "none" para Pydantic puro sin código de cliente.
• Familiaridad básica con FHIR y US Core (saber qué significan «perfil» y «slice» es suficiente).

Paso 1 — Generar Clases de Perfil

La generación de código se ejecuta a través de la herramienta Node, así que configure un pequeño proyecto generador junto a su aplicación Python:

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

Cree 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();

Los parámetros relevantes aquí:

• generateProfile: true — emite una clase envolvente por perfil con accesores tipados para extensiones, slices y valores fijos. Sin él solo se obtienen los modelos Pydantic R4 base.
• allowExtraFields: false — los modelos generados usan extra="forbid" de Pydantic, de modo que un campo desconocido lanza un error en tiempo de análisis en lugar de descartarse silenciosamente.
• primitiveTypeExtension: true — genera también los hermanos de extensión de tipo primitivo FHIR (los compañeros _field, p. ej. birthDateExtension) para poder adjuntar extensiones e ids a valores primitivos.
• treeShake: { ... } — solo se generan los canonicales listados y sus dependencias transitivas (~20 archivos en lugar de cientos).

Ejecútelo. prettyReport(report) imprime un resumen agrupado para ver qué se ha emitido sin tener que recorrer el directorio de salida:

$ npx tsx generate.ts
# generation logs omitted; this is the prettyReport summary
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 estructura en disco tiene este aspecto:

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

Apunte su aplicación Python al directorio fhir_types/ emitido e instale las dependencias:

pip install -r fhir_types/requirements.txt

El requirements.txt generado fija Pydantic y fhirpy, además de pytest y requests para las pruebas y los ejemplos.

El código completo del tutorial se encuentra en Aidbox/examples — generate.ts, load.py, avg.py, post.py, el CSV y el directorio fhir_types/ confirmado para que pueda explorar el código generado sin ejecutar el generador. Para una exploración más amplia de la API de perfiles, el repositorio de codegen también tiene un ejemplo de prueba python-r4-us-core. Ambos usan los nombres de atributo camelCase predeterminados, igual que los fragmentos aquí. (Pase fieldFormat: "snake_case" si prefiere escribir los atributos como birth_date, effective_date_time; la serialización siempre emite JSON camelCase correcto según FHIR de todas formas.)

Paso 2 — Fila a US Core Patient

La entrada es patients.csv — datos demográficos básicos más una lectura de TA por paciente. La raza usa los códigos de categoría OMB que US Core espera:

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 entrega cada fila como un dict[str, str] simple; el análisis numérico ocurre más tarde, cuando se pasan los valores a los setters de perfil tipados.

El perfil US Core Patient añade algunas extensiones y hace que identifier y name sean obligatorios. La clase generada tiene un setter tipado para cada uno:

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

Dos fases:

1. Construir el Patient simple — los campos requeridos por el perfil (identifier, name) y los de soporte obligatorio (gender, birthDate) como modelo Pydantic R4 tipado. Se construye con los nombres de atributo camelCase predeterminados (birthDate, los nombres de cable FHIR); los valores se validan inmediatamente (p. ej. gender es un Literal["male", "female", "other", "unknown"]).
2. Luego UscorePatientProfile.apply(base_patient) estampa meta.profile y devuelve una instancia de perfil con accesores tipados para las extensiones US Core. apply() envuelve el recurso en su lugar — el perfil muta el mismo objeto Patient.

Tres notas sobre lo que hace la API de perfil por usted:

• Tres formas de setter de extensión. set_race({ "ombCategory": ..., "text": ... }) acepta entrada plana — nótese que las claves de sub-extensión (ombCategory, detailed, text) son los nombres de slice en camelCase — y genera la fontanería de extension[] anidada. El mismo setter también acepta una instancia de perfil de extensión tipada (UscoreRaceExtension) o una Extension en bruto, y lanza un error si la url de una extensión en bruto no coincide.
• Las extensiones de valor único toman el valor directamente. us-core-individual-sex lleva un valueCoding, por lo que set_sex(Coding(code="female")) acepta un Coding (o una Extension en bruto).
• Sin setters para campos base de soporte obligatorio. gender, birthDate y address no están perfilados adicionalmente por US Core, por lo que la clase de perfil no emite envoltorios al estilo .set_gender() — rellénelos como campos normales de Patient. validate() aun así advertirá si falta un campo de soporte obligatorio.

Pydantic emite un UserWarning cuando una lista extension[] contiene dicts simples en lugar de instancias de Extension — algo esperado con la fontanería actual de dict plano. Silencie esto con warnings.filterwarnings("ignore", category=UserWarning, module="pydantic").

Paso 3 — Fila a US Core Blood Pressure

El perfil de TA es donde codegen realmente justifica su uso. El perfil US Core Blood Pressure:

• fija code a LOINC 85354-9 («Blood pressure panel»),
• fija un slice de categoría vital-signs,
• define los slices component[systolic] y component[diastolic] con discriminadores LOINC específicos (8480-6 y 8462-4),
• requiere un effectiveDateTime o effectivePeriod,
• requiere valueQuantity dentro de cada slice.

Enrollar eso a mano por fila es exactamente el tipo de tarea que codegen elimina. La clase generada lo reduce a tres setters:

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

Lo que ocurre entre bastidores:

• create() hace la ceremonia. Estampa meta.profile, rellena el code fijo (LOINC 85354-9), añade el slice de categoría vital-signs y agrega stubs vacíos de component[systolic] / component[diastolic] con los códigos discriminadores ya establecidos. create() acepta argumentos solo por nombre; create_resource() hace lo mismo pero devuelve una Observation simple en lugar de un envoltorio de perfil.
• set_systolic({ "value": ..., "unit": ... }) rellena el valueQuantity dentro del slice sistólico. El code discriminador en ese componente ya está ahí desde create() — solo hay que suministrar la lectura.
• validate() devuelve {"errors": [...], "warnings": [...]}. Los errores bloquean (campos obligatorios, campos excluidos, variantes de elección no permitidas, cardinalidad de slice). Las advertencias señalan problemas de soporte obligatorio. Una fila malformada falla rápido con el MRN — no se descubre en el momento del POST.

No se han escrito los códigos discriminadores. No se ha recordado 85354-9. Los setters encadenan fluidamente (cada uno devuelve el perfil), igual que la API de TypeScript.

Paso 4 — Ensamblar el Bundle

Cada fila produce un Patient y una Observation de TA vinculados por el marcador de posición urn:uuid del Patient. Se empaquetan como entradas de transacción. El Bundle y BundleEntry generados son genéricos sobre el recurso contenido, de modo que un Bundle[Patient | Observation] mantiene entry[].resource tipado a esa unión. (row_to_patient y row_to_bp son las funciones de los Pasos 2-3; en el ejemplo todas viven en un load.py, por lo que ya están en el ámbito aquí.)

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

Vale la pena destacar:

• to_resource() devuelve el modelo simple — el recurso Pydantic subyacente, sin envoltorio, listo para insertar en un BundleEntry.
• model_dump(by_alias=True, exclude_none=True) produce JSON FHIR — by_alias serializa a través de los alias del cable FHIR (de modo que una construcción en snake_case sigue emitiendo effectiveDateTime) y exclude_none descarta los campos con valor None. La llamada de serialización única que se usará en todas partes.
• Referencias urn:uuid. El fullUrl del paciente y el subject.reference de la observación comparten un UUID; el servidor lo resuelve a un id real al confirmar.

Paso 5 — Lectura: Media de TA desde el Bundle

Ahora léalo de vuelta. Analice bundle.json y calcule la media sistólica/diastólica para ejercitar la API del lado de lectura:

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)

Tres cosas que hace el perfil aquí:

• from_resource(obs) valida al envolver. Comprueba que meta.profile incluye la URL canónica y devuelve una instancia de perfil, lanzando un error si un recurso que declara el perfil está malformado — de modo que un bundle roto falla en tiempo de lectura, no en el siguiente acceso a un campo.
• Sin guardia de tipo incorporada. A diferencia del predicado is() de la API de TypeScript, las clases Python no incluyen una guardia estilo .filter(). Los candidatos se seleccionan manualmente — comprobando resourceType y canonical_url in meta.profile (véase arriba), o envolviendo from_resource() en try/except ValueError. De cualquier forma, canonical_url está expuesto como atributo de clase exactamente para esto.
• get_systolic() / get_diastolic() devuelven el valor plano del slice. No hay que recorrer component[].code.coding[].code para encontrar los códigos LOINC — el perfil ya sabe qué slice es cuál, y devuelve los datos de Quantity como un dict simple.

Eso es el ciclo completo: CSV → perfiles tipados → Bundle validado → lectura tipada de vuelta con getters conscientes del perfil. Las mismas pocas líneas procesarían TAs obtenidas de un servidor FHIR, cargadas desde un archivo, o recibidas en una Subscription — el perfil tipado es la forma común, independientemente de la fuente.

Paso 6 — Publicar el Bundle en un Servidor FHIR

El pipeline tipado es solo la mitad de la historia. Para ver realmente la confirmación de la transacción — IDs de paciente asignados, referencias urn:uuid reescritas, recursos almacenados y consultables — se necesita un servidor FHIR. Arranque y ejecute Aidbox:

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

Abra http://localhost:8080 en su navegador para obtener una licencia de desarrollador gratuita, luego extraiga el secreto del cliente raíz de docker-compose.yaml en una variable de entorno — reutilizada por los curls y el script Python a continuación:

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

Verifique que el endpoint FHIR está activo:

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

Debería ver un CapabilityStatement JSON.

Envíe el bundle.json que acaba de escribir con el cliente asíncrono 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 devuelve un bundle transaction-response — una entrada por entrada de entrada, cada una con 201 Created y una location que apunta al recurso almacenado:

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

Consulte una observación de vuelta y examine su 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..."

Sin urn:uuid — Aidbox reescribió los marcadores de posición de forma atómica al confirmar.

Verificación de Tipos del Pipeline

Los modelos generados son Pydantic v2, por lo que el conversor se puede verificar con mypy — ya fijado en el requirements.txt generado. Un requisito: habilite el plugin mypy de Pydantic (se incluye con Pydantic, sin instalación adicional), o mypy no podrá determinar que un Field(None, ...) predeterminado hace que un campo sea opcional e inundará con falsos errores de «argumento faltante» en cada modelo que construya.

Coloque un mypy.ini junto a su código:

[mypy]
strict = True
plugins = pydantic.mypy

Luego ejecútelo:

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

Tanto sus módulos de conversión — load.py, avg.py, post.py — como el paquete fhir_types/ generado resultan limpios. Las factorías tipadas, to_resource() y el genérico Bundle[Patient | Observation] se verifican correctamente, de modo que un tipo de campo incorrecto o un argumento requerido faltante se detectan antes de llegar al servidor. La capa de perfil generada también se verifica bajo --strict completo — el punto de que mypy.ini sean solo esas dos líneas: sin disable_error_code, sin strict_optional = False, nada editado a mano dentro de fhir_types/. Lo único que su propio código debe aportar es una guardia de None ordinaria al leer un campo opcional — el walrus en avg.py de arriba, o if entries is None: ... antes de indexar una lista. Eso es Python en modo estricto normal, no una peculiaridad del generador.

Hacia Dónde Ir a Continuación

• Más de la API de perfil. Otras factorías, getters y formas de slice/extensión se ejercitan en las pruebas de ejemplo de codegen: test_profile_patient.py, test_profile_bp.py, test_profile_bodyweight.py y test_profile_typed_bundle.py.
• Bundles tipados con slices de entrada con nombre. Un Bundle perfilado genera setters/getters por slice (set_patient_entry, get_organization_entry) con cardinalidad simple frente a ilimitada (max: *) gestionada automáticamente — véase la prueba de bundle tipado anterior.
• Ajustar la salida para su base de código. fieldFormat (snake_case/camelCase), client ("fhirpy"/"none"), allowExtraFields y primitiveTypeExtension son todos interruptores en .python({ ... }).
• Mezclar perfiles de múltiples paquetes. APIBuilder.fromPackage() encadena — US Core junto a su IG personalizado o una base regional. localStructureDefinitions() carga perfiles directamente desde una carpeta de JSON StructureDefinition.

Conclusión

El generador emite tanto los modelos Pydantic R4 base como una fina capa de clases de perfil encima — sin DSL en tiempo de ejecución, sin ORM, sin framework. to_resource() siempre devuelve un recurso Pydantic simple, y model_dump(by_alias=True, exclude_none=True) siempre devuelve JSON FHIR puro que puede enviarse a cualquier servidor.

@atomic-ehr/codegen tiene licencia MIT; los issues y PRs son bienvenidos.

GitHub | NPM | US Core IG