dhis2w-fhir-engine

What a FHIR-to-DHIS2 service installs: the facade library and routes, the R4 models, and FHIRPath, CQL, and ELM evaluation with clinical quality measures.

Build a middleware

A service of your own between another system and DHIS2, with no fhir.toml, no SUSHI, and no docker:

uv init myfacade && cd myfacade
uv add dhis2w-fhir-engine uvicorn

from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from pathlib import Path

from dhis2w_client import BasicAuth, Dhis2Client
from dhis2w_fhir_engine.facade import FacadeConfig, open_facade
from dhis2w_fhir_engine.facade.api import attach_facade, mount_facade
from fastapi import FastAPI

config = FacadeConfig(canonical="http://example.org/fhir/demo", root=Path(".facade"))


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
    auth = BasicAuth(username="admin", password="district")
    async with Dhis2Client(base_url="http://localhost:8080", auth=auth) as client:
        async with open_facade(config, client=client) as facade:
            attach_facade(app, facade)
            yield


app = FastAPI(lifespan=lifespan)
mount_facade(app, read_as_caller=True)

uv run uvicorn main:app

That serves /metadata, read and search over the forms and the people, $generate, and POST /QuestionnaireResponse, which sends each capture to DHIS2 while the caller waits, so they hear DHIS2's answer. read_as_caller has every caller present their own DHIS2 credentials, so DHIS2 applies their rights, and admin here is for a local demo instance only. Read Who can do what before anyone else can reach your service.

Build a middleware walks all of it, and examples/facade/ runs it in eleven steps.

Evaluate over FHIR data

The grammar, parser, AST, and evaluator layers know nothing about any particular FHIR release — FHIRPath is normative and CQL is 1.5, and neither names a version. Everything that does bind to a release lives under dhis2w_fhir_engine.r4 and reaches the neutral core as a value, so a later release lands as a sibling subpackage instead of a fork of the engine.

It also owns the R4 resource models at dhis2w_fhir_engine.r4.resources — Patient, Bundle, QuestionnaireResponse, Composition, Extension, and the rest. Every one is closed, frozen, and alias-aware, so model_dump_json(exclude_none=True, by_alias=True) reproduces the wire document key for key. The evaluators, the data sources, and the measure evaluator all accept either a wire dict or one of these models: a model is dumped once on entry and evaluation reads dicts from there on.

from dhis2w_fhir_engine import FHIRPathEvaluator
from dhis2w_fhir_engine.r4.resources import HumanName, Patient

FHIRPathEvaluator().evaluate("Patient.name.family", Patient(name=[HumanName(family="Kamara")]))
# ['Kamara']

The package depends on antlr4-python3-runtime, dhis2w-client, httpx2, and pydantic; the [cli] extra adds typer and rich. dhis2w_fhir_engine.evaluation is free of DHIS2 as code and evaluates expressions over FHIR-shaped JSON. dhis2w_fhir_engine.facade is what a FHIR-to-DHIS2 service is built from.

Install

uv add dhis2w-fhir-engine        # or: pip install dhis2w-fhir-engine

Quickstart — FHIRPath over a resource

from dhis2w_fhir_engine import FHIRPathEvaluator

patient = {
    "resourceType": "Patient",
    "active": True,
    "name": [{"given": ["Ada"], "family": "Lovelace"}],
    "birthDate": "1815-12-10",
}

evaluator = FHIRPathEvaluator()
print(evaluator.evaluate("Patient.name.given.first()", patient))
print(evaluator.evaluate_boolean("Patient.active and Patient.birthDate < @1900-01-01", patient))

Quickstart — CQL against a Bundle

retrieve expressions ([Patient], [Condition: "Diabetes"]) read through a data source. The R4 data sources index a Bundle or a list of resources and answer retrieves against it.

from dhis2w_fhir_engine import CQLEvaluator
from dhis2w_fhir_engine.r4 import BundleDataSource

bundle = {
    "resourceType": "Bundle",
    "type": "collection",
    "entry": [
        {"resource": {"resourceType": "Patient", "id": "p1", "birthDate": "1990-01-01"}},
        {"resource": {"resourceType": "Condition", "id": "c1", "subject": {"reference": "Patient/p1"}}},
    ],
}

evaluator = CQLEvaluator(data_source=BundleDataSource(bundle))
evaluator.compile("""
    library Example version '1.0'
    using FHIR version '4.0.1'

    define Patients: [Patient]
    define HasCondition: exists [Condition]
""")
print(evaluator.evaluate_definition("HasCondition"))

FHIRHelpers 4.0.1 is resolvable out of the box — include FHIRHelpers version '4.0.1' needs no library path. Pass include_builtins=False to leave it out.

Quickstart — a quality measure

from dhis2w_fhir_engine.r4 import BundleDataSource, MeasureEvaluator

evaluator = MeasureEvaluator(data_source=BundleDataSource(bundle))
evaluator.load_measure("""
    library CoverageMeasure version '1.0'
    using FHIR version '4.0.1'
    context Patient

    define "Initial Population": true
    define "Denominator": "Initial Population"
    define "Numerator": exists [Observation]
""")

report = evaluator.evaluate_population(patients)
print(report.groups[0].populations["numerator"].count)
print(report.to_fhir())  # a FHIR R4 MeasureReport

MeasureEvaluator finds each population by the conventional CQL definition name (Initial Population, Denominator, Numerator, the exclusion and exception variants), or you can declare the groups yourself with add_population and add_stratifier. evaluate_patient returns one patient's population membership; evaluate_population aggregates and computes the measure score.

ELM

The engine reads and writes ELM, the standardised JSON form of a compiled CQL library, so libraries compiled elsewhere run here and libraries written here export for other implementations.

from dhis2w_fhir_engine import ELMEvaluator, ELMLoader, ELMSerializer

library = ELMLoader().load_file("library.json")  # run ELM produced by another compiler
result = ELMEvaluator().evaluate_definition("Numerator")

elm_json = ELMSerializer().serialize_library_json(cql_source)  # export CQL as ELM

Command line

The console script d2w-fhir-engine mounts three sub-apps over the same engine. It needs the command line extra: uv add 'dhis2w-fhir-engine[cli]'.

d2w-fhir-engine fhirpath eval "Patient.name.given" --resource patient.json
d2w-fhir-engine fhirpath parse-file expressions.txt
d2w-fhir-engine fhirpath repl

d2w-fhir-engine cql eval "1 + 2 * 3"
d2w-fhir-engine cql run library.cql --data bundle.json
d2w-fhir-engine cql check library.cql
d2w-fhir-engine cql measure measure.cql --data bundle.json
d2w-fhir-engine cql export library.cql --output library.elm.json
d2w-fhir-engine cql repl

d2w-fhir-engine elm load library.elm.json
d2w-fhir-engine elm run library.elm.json --define Numerator
d2w-fhir-engine elm convert library.cql --output library.elm.json

Each sub-app also carries parse, ast, tokens, show, and validate for inspecting a source before running it.

--data reads by one rule wherever it appears: a Bundle becomes the data source retrieves read, any other resource becomes the context resource the evaluation is about. cql measure takes both halves off a Bundle - every Patient entry is a person to evaluate, and the whole Bundle is what the numerator retrieves from.

FHIR version binding

FhirVersionBinding is the whole of the engine's version knowledge: the patient-reference element paths per resource type, the canonical base profile URLs conformsTo() compares against, and the built-in CQL libraries. dhis2w_fhir_engine.r4 builds one for R4 and importing the package installs it as the default, so nothing has to be passed for R4 work.

from dhis2w_fhir_engine import CQLEvaluator, FhirVersionBinding
from dhis2w_fhir_engine.r4 import R4_BINDING

CQLEvaluator(fhir_binding=R4_BINDING)  # explicit; the same binding is the default

An evaluator receives its binding as a value and never imports a version subpackage, so a release subpackage only has to export a FhirVersionBinding and, where the wire shapes differ, its own data sources and measure report writer.

Layout

dhis2w_fhir_engine/
  binding.py        FhirVersionBinding and the default-binding registry
  cli/              one Typer app: fhirpath, cql, elm
  grammars/         the HL7 cql.g4 and fhirpath.g4 grammars
  generated/        ANTLR output for both grammars (excluded from every linter)
  engine/           version-neutral: context, types, functions, fhirpath/, cql/, elm/, units/
  r4/               FHIR R4: binding, data sources, measure evaluation, terminology, FHIRHelpers

Tests

cd packages/dhis2w-fhir-engine
uv run pytest -q                        # unit suite plus the HL7 CQL and FHIRPath compliance suites
uv run pytest -q tests/e2e_dhis2 -m slow   # end-to-end against a running local DHIS2 stack

The compliance suites under tests/compliance/ run the official HL7 test XML for CQL and for FHIRPath R4. The tests/e2e_dhis2/ group evaluates FHIRPath, CQL retrieves, and a measure over FHIR resources built from seeded DHIS2 data, and skips with a stated reason when the stack is down.