Modeling a patient roster as a FHIR Group (Da Vinci ATR)

Part 2 of 7 — Roster API Series: migrating member-file ingestion from CSV over SFTP to a FHIR API.

In Part 1 the conclusion was that a roster CSV can’t express identity namespaces, intent, or change. Before building any endpoint, the data needs a model that can. HL7’s Da Vinci Member Attribution (ATR) List IG gives us one, built on three FHIR R4 resources:

• Group — the list itself, the contract it belongs to, and its lifecycle status.
• Patient — one per member, referenced from Group.member.entity.
• Coverage — which plan/program the member is attributed under.

Plus Organization/Practitioner for the payer and the attributed provider.

The Group, element by element

The ATR Group profile pins down what an attribution list is:

And per member:

Two things here fix Part 1 problems directly. attributionListStatus gives the list a lifecycle — it can be reconciled while draft/open and frozen when final. And changeType plus inactive make change part of the payload instead of something you diff for yourself.

Identity: stop sending bare MRNs

The single biggest modeling win is Identifier.system. An MRN becomes a pair:

{
  "system": "https://ehr.example-clinic.org/fhir/sid/mrn",
  "value": "00123456"
}

Now two EHRs can both have MRN 00123456 without colliding. Payer member IDs get their own system. You can carry several identifiers per Patient, which is exactly what you need when the payer knows the member by member ID and the clinic by MRN.

For the name + DOB fallback, FHIR doesn’t make the matching problem disappear — but it moves it into an explicit operation (Patient/$match) that returns candidates with a score, instead of a silent join in a vendor’s batch job.

The CSV we start from

Synthetic, of course:

mrn,first_name,last_name,dob,plan_type,provider_npi,period_start,period_end
00123456,Maria,Gomez,1958-03-07,MA,1234567893,2026-01-01,2026-12-31
00123457,John,Smith,1949-11-21,ACO,1234567893,2026-01-01,2026-12-31

Converter: CSV → ATR Group bundle

I keep the models thin with Pydantic and emit plain dicts — no FHIR library dependency needed for the first iteration.

# roster_to_fhir.py
import csv
import json
import uuid
from datetime import date

from pydantic import BaseModel, field_validator

ATR = "http://hl7.org/fhir/us/davinci-atr/StructureDefinition"
MRN_SYSTEM = "https://ehr.example-clinic.org/fhir/sid/mrn"
NPI_SYSTEM = "http://hl7.org/fhir/sid/us-npi"


class RosterRow(BaseModel):
    mrn: str
    first_name: str
    last_name: str
    dob: date
    plan_type: str
    provider_npi: str
    period_start: date
    period_end: date

    @field_validator("mrn", "provider_npi")
    @classmethod
    def not_blank(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("required")
        return v.strip()


def patient(row: RosterRow) -> dict:
    return {
        "resourceType": "Patient",
        "id": str(uuid.uuid5(uuid.NAMESPACE_URL, f"{MRN_SYSTEM}|{row.mrn}")),
        "identifier": [{"system": MRN_SYSTEM, "value": row.mrn}],
        "name": [{"family": row.last_name, "given": [row.first_name]}],
        "birthDate": row.dob.isoformat(),
    }


def coverage(row: RosterRow, patient_id: str) -> dict:
    return {
        "resourceType": "Coverage",
        "id": f"cov-{patient_id}",
        "status": "active",
        "beneficiary": {"reference": f"Patient/{patient_id}"},
        "payor": [{"reference": "Organization/payer"}],
        "class": [{
            "type": {"coding": [{
                "system": "http://terminology.hl7.org/CodeSystem/coverage-class",
                "code": "plan",
            }]},
            "value": row.plan_type,
        }],
    }


def member(row: RosterRow, patient_id: str, cov_id: str) -> dict:
    return {
        "extension": [
            {"url": f"{ATR}/ext-changeType", "valueCode": "new"},
            {"url": f"{ATR}/ext-coverageReference",
             "valueReference": {"reference": f"Coverage/{cov_id}"}},
            {"url": f"{ATR}/ext-attributedProvider",
             "valueReference": {"identifier": {"system": NPI_SYSTEM,
                                                "value": row.provider_npi}}},
        ],
        "entity": {"reference": f"Patient/{patient_id}"},
        "period": {"start": row.period_start.isoformat(),
                   "end": row.period_end.isoformat()},
        "inactive": False,
    }


def convert(path: str) -> tuple[dict, list[dict]]:
    resources, members, errors = [], [], []
    with open(path, newline="") as f:
        for n, raw in enumerate(csv.DictReader(f), start=2):
            try:
                row = RosterRow(**raw)
            except Exception as exc:  # row-level feedback, not a silent drop
                errors.append({"line": n, "error": str(exc)})
                continue
            p = patient(row)
            c = coverage(row, p["id"])
            resources += [p, c]
            members.append(member(row, p["id"], c["id"]))

    group = {
        "resourceType": "Group",
        "id": "roster-2026",
        "extension": [
            {"url": f"{ATR}/ext-attributionListStatus", "valueCode": "draft"},
            {"url": f"{ATR}/ext-contractValidityPeriod",
             "valuePeriod": {"start": "2026-01-01", "end": "2026-12-31"}},
        ],
        "identifier": [{"system": NPI_SYSTEM, "value": "1234567893"}],
        "type": "person",
        "actual": True,
        "managingEntity": {"reference": "Organization/payer"},
        "quantity": len(members),
        "member": members,
    }
    bundle = {
        "resourceType": "Bundle",
        "type": "collection",
        "entry": [{"resource": r} for r in [group, *resources]],
    }
    return bundle, errors


if __name__ == "__main__":
    bundle, errors = convert("roster.csv")
    json.dump(bundle, open("roster.bundle.json", "w"), indent=2)
    print(f"{len(bundle['entry'])} resources, {len(errors)} rejected rows")
    for e in errors:
        print(e)

Notes on decisions:

• Deterministic Patient IDs (uuid5 over system|value). Re-running the converter on the same CSV yields the same IDs, so later diffs compare like with like.
• Row-level errors are output, not logs. This is the acknowledgement channel the SFTP flow never had. In Part 3 it becomes an OperationOutcome.
• Provider by identifier, not reference. At conversion time we may not have a Practitioner resource; a logical reference by NPI is valid FHIR and resolvable later.
• Plan type lives on Coverage, not on Group. One Group per contract; the program a member falls under is a property of their coverage.

Validate before you trust it

Run the output through the official HL7 validator with the ATR package loaded:

$ java -jar validator_cli.jar roster.bundle.json \
    -version 4.0.1 \
    -ig hl7.fhir.us.davinci-atr#2.1.0

Expect complaints the first time — the ATR Patient and Coverage profiles inherit US Core constraints (e.g. gender must be present on Patient). That’s the point: the validator is a contract test you couldn’t run against a CSV.

What we have now

A roster that carries namespaced identity, a lifecycle status, a contract period, per-member attribution periods and an explicit change type. Still a static file — but one an API can accept, validate and answer about.

Next: Part 3 — From full replace to deltas, where this Group gets $member-add and $member-remove endpoints in FastAPI.

References

• ATR Group profile
• ATR change type codes
• ATR attribution list status codes
• ATR background — data model
• FHIR R4 Group
• FHIR R4 Coverage
• FHIR R4 Patient/$match
• HL7 FHIR Validator
• Pydantic validators

26 Oct 2026 · #fhir #healthcare #python #datamodeling #rosterapiseries

GLB_