From full replace to deltas: $member-add and $member-remove in FastAPI

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

The most dangerous rule in a flat-file roster is last file wins. A truncated export becomes a mass disenrollment and nobody gets an error. This post replaces that rule with explicit operations, using the ones the Da Vinci ATR IG already defines:

The rules the IG gives us

From the ATR formal spec:

• $member-add must support these parameter combinations: memberId + providerNpi, the same plus attributionPeriod, patientReference + providerReference, and the same plus attributionPeriod.
• $member-remove accepts memberId, memberId + providerNpi, patientReference, patientReference + providerReference, or that plus coverageReference.
• Both are only allowed while the list status is draft or open. On a final list the server rejects the call with an OperationOutcome.
• A removed member is not deleted: Group.member.inactive = true and ext-changeType = dropped both mean “no longer attributed”.
• Any change to members, periods, coverage or providers creates a new Group version.

That last point is the audit trail that didn’t exist with CSV snapshots.

Inactive, not deleted

Why not just drop the entry from Group.member?

• Billing and quality need history. “Who was attributed on March 15?” must stay answerable after the member leaves.
• Removal is a fact with a reason. dropped (left attribution) and cancelled (added by mistake in the previous version) are different events with different downstream consequences.
• Idempotency. Removing an already-inactive member is a no-op you can detect; removing an entry that isn’t there is ambiguous.

The service

In-memory store to keep the focus on semantics. Swap for Postgres with a group_versions table when you build it for real.

# app.py
from copy import deepcopy
from datetime import datetime, timezone

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse

ATR = "http://hl7.org/fhir/us/davinci-atr/StructureDefinition"
NPI = "http://hl7.org/fhir/sid/us-npi"
FHIR_JSON = "application/fhir+json"

app = FastAPI(title="Roster API")

GROUPS: dict[str, list[dict]] = {}   # group id -> list of versions (append-only)


# ---------- helpers ----------
def outcome(status: int, severity: str, code: str, text: str) -> JSONResponse:
    return JSONResponse(
        status_code=status,
        media_type=FHIR_JSON,
        content={"resourceType": "OperationOutcome",
                 "issue": [{"severity": severity, "code": code,
                            "diagnostics": text}]},
    )


def ext(resource: dict, name: str):
    for e in resource.get("extension", []):
        if e["url"] == f"{ATR}/{name}":
            return e
    return None


def current(gid: str) -> dict:
    if gid not in GROUPS:
        raise HTTPException(404, "Group not found")
    return GROUPS[gid][-1]


def new_version(gid: str, group: dict) -> dict:
    g = deepcopy(group)
    vid = len(GROUPS[gid]) + 1
    g["meta"] = {"versionId": str(vid),
                 "lastUpdated": datetime.now(timezone.utc).isoformat()}
    g["quantity"] = sum(1 for m in g["member"] if not m.get("inactive"))
    GROUPS[gid].append(g)
    return g


def params(body: dict) -> dict:
    """Flatten a FHIR Parameters resource into {name: value}."""
    out = {}
    for p in body.get("parameter", []):
        value = next((v for k, v in p.items() if k.startswith("value")), None)
        out[p["name"]] = value
    return out


def find_member(group: dict, p: dict) -> dict | None:
    for m in group["member"]:
        ref = m["entity"].get("reference")
        if p.get("patientReference") and ref == p["patientReference"]["reference"]:
            return m
        if p.get("memberId") and m.get("_memberId") == p["memberId"]["value"]:
            return m
    return None


def require_open(group: dict):
    status = ext(group, "ext-attributionListStatus")["valueCode"]
    if status not in ("draft", "open"):
        raise HTTPException(409, f"List status is '{status}'; changes not allowed")


# ---------- operations ----------
@app.post("/Group/{gid}/$member-add")
def member_add(gid: str, body: dict):
    group = deepcopy(current(gid))
    try:
        require_open(group)
    except HTTPException as e:
        return outcome(e.status_code, "error", "business-rule", e.detail)

    p = params(body)
    by_id = p.get("memberId") and p.get("providerNpi")
    by_ref = p.get("patientReference") and p.get("providerReference")
    if not (by_id or by_ref):
        return outcome(400, "error", "required",
                       "Need memberId+providerNpi or patientReference+providerReference")

    provider = (p["providerReference"] if by_ref
                else {"identifier": {"system": NPI, "value": p["providerNpi"]["value"]}})
    existing = find_member(group, p)

    if existing and not existing.get("inactive"):
        return outcome(200, "information", "informational", "Member already active")

    entry = existing or {
        "entity": p.get("patientReference")
                  or {"identifier": p["memberId"]},
        "_memberId": p["memberId"]["value"] if p.get("memberId") else None,
    }
    entry["inactive"] = False
    entry["extension"] = [
        {"url": f"{ATR}/ext-changeType",
         "valueCode": "changed" if existing else "new"},
        {"url": f"{ATR}/ext-attributedProvider", "valueReference": provider},
    ]
    if p.get("attributionPeriod"):
        entry["period"] = p["attributionPeriod"]
    if not existing:
        group["member"].append(entry)

    return JSONResponse(new_version(gid, group), media_type=FHIR_JSON)


@app.post("/Group/{gid}/$member-remove")
def member_remove(gid: str, body: dict):
    group = deepcopy(current(gid))
    try:
        require_open(group)
    except HTTPException as e:
        return outcome(e.status_code, "error", "business-rule", e.detail)

    p = params(body)
    m = find_member(group, p)
    if m is None:
        return outcome(404, "error", "not-found", "Member not in list")
    if m.get("inactive"):
        return outcome(200, "information", "informational", "Already inactive")

    m["inactive"] = True
    m["extension"] = [e for e in m.get("extension", [])
                      if not e["url"].endswith("ext-changeType")]
    m["extension"].append({"url": f"{ATR}/ext-changeType", "valueCode": "dropped"})
    end = datetime.now(timezone.utc).date().isoformat()
    m.setdefault("period", {})["end"] = end

    return JSONResponse(new_version(gid, group), media_type=FHIR_JSON)


@app.post("/Group/{gid}/$confirm-attribution-list")
def confirm(gid: str):
    group = deepcopy(current(gid))
    ext(group, "ext-attributionListStatus")["valueCode"] = "final"
    new_version(gid, group)
    return JSONResponse(status_code=202, content=None)


@app.get("/Group/{gid}/_history/{vid}")
def read_version(gid: str, vid: int):
    versions = GROUPS.get(gid, [])
    if not 1 <= vid <= len(versions):
        raise HTTPException(404)
    return JSONResponse(versions[vid - 1], media_type=FHIR_JSON)

A few things I want to call out:

• Append-only versions. GROUPS[gid] is a list. Every mutation creates a version with meta.versionId; reading _history/{vid} answers “what did the list look like then”. That’s the audit trail.
• Idempotent by design. Adding an active member or removing an inactive one returns an informational OperationOutcome instead of creating a version. Retries are safe.
• Status gating returns OperationOutcome. The IG requires it; it’s also your row-level acknowledgement.
• _memberId is a private shortcut for the demo. In a real store, index identifiers properly instead of stuffing fields into the resource.
• About $confirm-attribution-list: the IG text says the server returns “201 Accepted”. 201 is Created; 202 is Accepted. I return 202 and flag it — check the published version you target.

Calling it

$ curl -s -X POST localhost:8000/Group/roster-2026/\$member-add \
    -H 'Content-Type: application/fhir+json' \
    -d '{
      "resourceType": "Parameters",
      "parameter": [
        {"name": "memberId",
         "valueIdentifier": {"system": "https://payer.example.org/member-id", "value": "M-7781"}},
        {"name": "providerNpi",
         "valueIdentifier": {"system": "http://hl7.org/fhir/sid/us-npi", "value": "1234567893"}},
        {"name": "attributionPeriod",
         "valuePeriod": {"start": "2026-11-01", "end": "2026-12-31"}}
      ]}' | jq '.meta.versionId, .quantity'

Migrating a nightly CSV onto deltas

You won’t get every upstream system to call $member-add on day one. The bridge: keep receiving full snapshots, but turn them into deltas on your side, then apply them through the API.

def snapshot_to_ops(prev_ids: set[str], curr_ids: set[str], max_drop_ratio=0.05):
    added, removed = curr_ids - prev_ids, prev_ids - curr_ids
    if prev_ids and len(removed) / len(prev_ids) > max_drop_ratio:
        raise RuntimeError(
            f"{len(removed)} removals ({len(removed)/len(prev_ids):.0%}) "
            "exceeds threshold — refusing to apply. Truncated export?")
    return [("add", i) for i in added] + [("remove", i) for i in removed]

That threshold is the single cheapest guard against the truncated-file disaster from Part 1. With full replace there’s no place to put it; with deltas it’s three lines.

Next: Part 4 — Bulk export and async jobs, because consumers still need the whole list, and a 200k-member Group doesn’t fit in one JSON response.

References

• ATR formal specification (operations, status rules, versioning)
• OperationDefinition: $member-add
• OperationDefinition: $member-remove
• ATR change type codes
• FHIR R4 Operations framework
• FHIR R4 OperationOutcome
• FHIR R4 versioning and _history
• RFC 9110 — 201 Created / 202 Accepted
• FastAPI — Path parameters

27 Oct 2026 · #fhir #fastapi #python #apis #rosterapiseries

GLB_