EHR

openbtk.data.ehr loads FHIR R4, OMOP and HL7 v2 into one PatientRecord schema. It must be imported explicitly (it registers loader.ehr.fhir, loader.ehr.omop and loader.ehr.hl7v2). Reading needs the ehr extra: pip install "openbtk[ehr]". See the EHR guide for a worked example.

Schemas

openbtk.data.ehr.schemas.PatientRecord

Bases: BaseModel

One patient's structured clinical record, source-agnostic.

Example

record = PatientRecord( ... patient_id="pt-1", ... demographics=Demographics(gender="female"), ... source_system="fhir-r4", ... ) record.deid_status

openbtk.data.ehr.schemas.Demographics

Bases: BaseModel

A patient's demographic attributes, as loaded -- before or after de-identification (PatientRecord.deid_status governs that, not this type).

race/ethnicity are free text here, not an enum: FHIR's US Core extensions and OMOP's race_concept_id/ethnicity_concept_id both ultimately resolve to a categorical value from their own vocabularies (OMB categories, in practice), but Openbtk does not bundle either vocabulary (docs/09_CODING_STANDARDS.md section 7 -- openbtk.terminology is the right home for that resolution, and it is not built yet, M7). Loaders populate the human-readable display text they already have rather than inventing a premature enum.

Example

Demographics(gender="female").gender 'female'

openbtk.data.ehr.schemas.Encounter

Bases: BaseModel

A single care encounter (visit, admission, ...).

encounter_id is the join key CodedEvent.encounter_ref and Measurement.encounter_ref point back to -- loaders are responsible for using the same, bare identifier on both sides (docs/05_DATA_MODALITY_SPEC.md section 2.1's own worked timeline example shows every event carrying its encounter inline for exactly this reason).

Example

Encounter(encounter_id="enc-1", encounter_type="inpatient").encounter_id 'enc-1'

openbtk.data.ehr.schemas.CodedEvent

Bases: BaseModel

A single coded clinical event: a condition, medication or procedure.

Example

from openbtk.core.schemas import CodeSystem event = CodedEvent(code="385093006", system=CodeSystem.SNOMED) event.system.value 'SNOMED'

openbtk.data.ehr.schemas.Measurement

Bases: BaseModel

A single observation or lab result.

value is float | str | None because not every observation is numeric -- a qualitative result ("Positive", "Detected") is real, common lab data, not an edge case to reject.

display is a deliberate, disclosed addition beyond docs/05_DATA_MODALITY_SPEC.md section 2.1's original field list (which otherwise matches exactly) -- PatientTimelineSerializer (task 6.5) needs a human-readable name ("WBC") to render a lab result at all; a bare LOINC code is not something a clinician or an LLM reading a rendered timeline can interpret. CodedEvent already had this field; Measurement lacking it was an oversight in the original spec, fixed here rather than shipping a serializer that can only ever show raw codes.

Example

m = Measurement(code="6690-2", display="WBC", value=14.2, unit="10*3/uL") m.system.value 'LOINC'

Loaders

openbtk.data.ehr.fhir.FHIRLoader

Bases: BaseLoader[str, PatientRecord]

Load one PatientRecord per FHIR R4 Bundle JSON file in a directory. See this module's own docstring for the loader's exact, disclosed scope.

Example

import contextlib, io, json, tempfile from pathlib import Path bundle = { ... "resourceType": "Bundle", ... "type": "collection", ... "entry": [{"resource": { ... "resourceType": "Patient", "id": "pt-1", "gender": "female", ... }}], ... } with tempfile.TemporaryDirectory() as d: ... _ = (Path(d) / "pt-1.json").write_text(json.dumps(bundle)) ... with contextlib.redirect_stdout(io.StringIO()): ... records = list(FHIRLoader().load(d)) records[0].patient_id 'pt-1'

load

load(source: str) -> Iterator[PatientRecord]

LoaderError: If source is not a directory, a file is not valid JSON, is not a FHIR Bundle, or a bundle contains no Patient resource.

Source code in src/openbtk/data/ehr/fhir.py

def load(self, source: str) -> Iterator[PatientRecord]:
    """Raises:
    LoaderError: If ``source`` is not a directory, a file is not
        valid JSON, is not a FHIR ``Bundle``, or a bundle contains no
        ``Patient`` resource.
    """
    directory = Path(source)
    if not directory.is_dir():
        raise LoaderError(
            f"Not a directory: {source}",
            context={"modality": "ehr", "stage": "load"},
        )
    fhir_r4b = require("fhir.resources.R4B", extra="ehr")
    log.info("loader.start", modality="ehr", source=str(directory))
    for path in sorted(directory.glob("*.json")):
        yield self._record_from_bundle_file(path, fhir_r4b)

openbtk.data.ehr.omop.OMOPLoader

OMOPLoader(*, batch_size: int = 10000, condition_system: CodeSystem = CodeSystem.SNOMED, drug_system: CodeSystem = CodeSystem.RXNORM, procedure_system: CodeSystem = CodeSystem.CPT, measurement_system: CodeSystem = CodeSystem.LOINC)

Bases: BaseLoader[str, PatientRecord]

Load PatientRecords from a directory of OMOP CDM v5.4 Parquet tables (person.parquet required; visit_occurrence, condition_occurrence, drug_exposure, procedure_occurrence and measurement are each read if present, skipped if absent). See this module's own docstring for real, disclosed scope and memory trade-offs.

Example

import contextlib, io, tempfile from pathlib import Path import pyarrow as pa import pyarrow.parquet as pq with tempfile.TemporaryDirectory() as d: ... table = pa.table({ ... "person_id": [1], ... "gender_concept_id": [8532], ... "year_of_birth": [1980], ... "month_of_birth": [5], ... "day_of_birth": [1], ... }) ... _ = pq.write_table(table, Path(d) / "person.parquet") ... with contextlib.redirect_stdout(io.StringIO()): ... records = list(OMOPLoader().load(d)) records[0].patient_id '1' records[0].demographics.gender 'female'

Source code in src/openbtk/data/ehr/omop.py

def __init__(
    self,
    *,
    batch_size: int = 10_000,
    condition_system: CodeSystem = CodeSystem.SNOMED,
    drug_system: CodeSystem = CodeSystem.RXNORM,
    procedure_system: CodeSystem = CodeSystem.CPT,
    measurement_system: CodeSystem = CodeSystem.LOINC,
) -> None:
    self._batch_size = batch_size
    self._condition_system = condition_system
    self._drug_system = drug_system
    self._procedure_system = procedure_system
    self._measurement_system = measurement_system

load

load(source: str) -> Iterator[PatientRecord]

LoaderError: If source is not a directory, lacks person.parquet, or a Parquet table fails to parse.

Source code in src/openbtk/data/ehr/omop.py

def load(self, source: str) -> Iterator[PatientRecord]:
    """Raises:
    LoaderError: If ``source`` is not a directory, lacks
        ``person.parquet``, or a Parquet table fails to parse.
    """
    directory = Path(source)
    if not directory.is_dir():
        raise LoaderError(
            f"Not a directory: {source}",
            context={"modality": "ehr", "stage": "load"},
        )
    person_path = directory / "person.parquet"
    if not person_path.is_file():
        raise LoaderError(
            f"{source} is missing the required person.parquet table.",
            context={"modality": "ehr", "stage": "load"},
        )
    pq = require("pyarrow.parquet", extra="ehr")
    log.info("loader.start", modality="ehr", source=str(directory))

    encounters_by_person = self._load_visits(directory, pq)
    conditions_by_person = self._load_events(
        directory, "condition_occurrence.parquet", pq, self._condition_row_to_event
    )
    medications_by_person = self._load_events(
        directory, "drug_exposure.parquet", pq, self._drug_row_to_event
    )
    procedures_by_person = self._load_events(
        directory, "procedure_occurrence.parquet", pq, self._procedure_row_to_event
    )
    observations_by_person = self._load_measurements(directory, pq)

    for row in self._iter_rows(person_path, pq):
        person_id = str(row["person_id"])
        yield PatientRecord(
            patient_id=person_id,
            demographics=self._demographics_from_row(row),
            encounters=encounters_by_person.get(person_id, []),
            conditions=conditions_by_person.get(person_id, []),
            medications=medications_by_person.get(person_id, []),
            procedures=procedures_by_person.get(person_id, []),
            observations=observations_by_person.get(person_id, []),
            source_system=_SOURCE_SYSTEM,
        )

openbtk.data.ehr.hl7v2.HL7v2Loader

HL7v2Loader(*, default_utc_offset_hours: float = 0.0, encoding: str = 'utf-8')

Bases: BaseLoader[str, PatientRecord]

Load PatientRecords from HL7 v2 messages. See the module docstring for the exact scope.

Parameters:

Name	Type	Description	Default
default_utc_offset_hours	float	The offset assumed for a timestamp that carries none.	0.0
encoding	str	The text encoding of the files.	'utf-8'

Example

import contextlib, io, tempfile from pathlib import Path message = "\r".join([ ... "MSH|^~\&|APP|FAC|RCV|FAC|20240314101500||ADT^A01|M1|P|2.5", ... "PID|1||PT0001^^^HOSP^MR||DOE^JANE||19800601|F", ... ]) with tempfile.TemporaryDirectory() as d: ... _ = (Path(d) / "a.hl7").write_text(message) ... with contextlib.redirect_stdout(io.StringIO()): ... records = list(HL7v2Loader().load(d)) (records[0].patient_id, records[0].demographics.gender) ('PT0001', 'female')

Source code in src/openbtk/data/ehr/hl7v2.py

def __init__(
    self, *, default_utc_offset_hours: float = 0.0, encoding: str = "utf-8"
) -> None:
    self._offset = default_utc_offset_hours
    self._encoding = encoding

load

load(source: str) -> Iterator[PatientRecord]

LoaderError: If source is neither a file nor a directory, a file cannot be read, or a message cannot be parsed. The message names the file and the message's position, never its content.

Source code in src/openbtk/data/ehr/hl7v2.py

def load(self, source: str) -> Iterator[PatientRecord]:
    """Raises:
    LoaderError: If ``source`` is neither a file nor a directory, a file cannot
        be read, or a message cannot be parsed. The message names the file and
        the message's position, never its content.
    """
    path = Path(source)
    if path.is_dir():
        files = sorted(p for p in path.iterdir() if p.suffix.lower() == ".hl7")
    elif path.is_file():
        files = [path]
    else:
        raise LoaderError(
            f"Not a file or directory: {source}",
            context={"modality": "ehr", "stage": "load"},
        )
    parser = require("hl7apy.parser", extra="ehr")
    consts = require("hl7apy.consts", extra="ehr")
    log.info("loader.start", modality="ehr", format="hl7v2", source=str(path))
    for file in files:
        yield from self._patients_in(file, parser, consts)

Timelines and text

openbtk.data.ehr.temporal.TemporalNormalizer

TemporalNormalizer(*, anchor_missing_timestamps: bool = True)

Bases: BasePreprocessor[PatientRecord]

Anchor missing event timestamps to their encounter and sort every event list chronologically.

Parameters:

Name	Type	Description	Default
anchor_missing_timestamps	bool	When True (default), an event with no timestamp of its own but a resolvable encounter_ref borrows that encounter's start time.	True

Example

from datetime import datetime, UTC from openbtk.core.schemas import CodeSystem from openbtk.data.ehr.schemas import ( ... CodedEvent, Demographics, Encounter, PatientRecord, ... ) record = PatientRecord( ... patient_id="pt-1", ... demographics=Demographics(), ... encounters=[Encounter( ... encounter_id="enc-1", ... start=datetime(2024, 3, 14, 8, tzinfo=UTC), ... )], ... conditions=[CodedEvent( ... code="385093006", system=CodeSystem.SNOMED, encounter_ref="enc-1", ... )], ... source_system="fhir-r4", ... ) normalized = TemporalNormalizer().process(record) normalized.conditions[0].timestamp datetime.datetime(2024, 3, 14, 8, 0, tzinfo=datetime.timezone.utc)

Source code in src/openbtk/data/ehr/temporal.py

def __init__(self, *, anchor_missing_timestamps: bool = True) -> None:
    self._anchor_missing_timestamps = anchor_missing_timestamps

openbtk.pipelines.timeline.PatientTimelineSerializer

PatientTimelineSerializer(*, encounter_label: str = 'Encounter', condition_label: str = 'Condition', medication_label: str = 'Medication', procedure_label: str = 'Procedure', observation_label: str = 'Lab', encounter_template: Callable[[Encounter], str] = _default_encounter_template, condition_template: Callable[[CodedEvent], str] = _default_coded_event_template, medication_template: Callable[[CodedEvent], str] = _default_coded_event_template, procedure_template: Callable[[CodedEvent], str] = _default_coded_event_template, observation_template: Callable[[Measurement], str] = _default_observation_template, source: str = 'ehr-timeline')

Render a PatientRecord as a chronological plain-text timeline.

Parameters:

Name	Type	Description	Default
encounter_label	str	Column label for encounter rows.	'Encounter'
condition_label	str	Column label for condition rows.	'Condition'
medication_label	str	Column label for medication rows.	'Medication'
procedure_label	str	Column label for procedure rows.	'Procedure'
observation_label	str	Column label for observation/lab rows.	'Lab'
encounter_template	Callable[[Encounter], str]	Overrides the default one-line encounter description. Same idea for the other four *_template args.	_default_encounter_template
source	str	ClinicalTextRecord.source value for every record this serializer produces.	'ehr-timeline'

Example

from datetime import datetime, UTC from openbtk.core.schemas import CodeSystem from openbtk.data.ehr.schemas import CodedEvent, Demographics, PatientRecord record = PatientRecord( ... patient_id="pt-1", ... demographics=Demographics(), ... conditions=[CodedEvent( ... code="385093006", system=CodeSystem.SNOMED, ... display="Community-acquired pneumonia", ... timestamp=datetime(2024, 3, 14, tzinfo=UTC), ... )], ... source_system="fhir-r4", ... ) text_record = PatientTimelineSerializer().serialize(record) "Community-acquired pneumonia (SNOMED 385093006)" in text_record.text True text_record.patient_ref 'pt-1'

Source code in src/openbtk/pipelines/timeline.py

def __init__(
    self,
    *,
    encounter_label: str = "Encounter",
    condition_label: str = "Condition",
    medication_label: str = "Medication",
    procedure_label: str = "Procedure",
    observation_label: str = "Lab",
    encounter_template: Callable[[Encounter], str] = _default_encounter_template,
    condition_template: Callable[[CodedEvent], str] = _default_coded_event_template,
    medication_template: Callable[
        [CodedEvent], str
    ] = _default_coded_event_template,
    procedure_template: Callable[[CodedEvent], str] = _default_coded_event_template,
    observation_template: Callable[
        [Measurement], str
    ] = _default_observation_template,
    source: str = "ehr-timeline",
) -> None:
    self._encounter_label = encounter_label
    self._condition_label = condition_label
    self._medication_label = medication_label
    self._procedure_label = procedure_label
    self._observation_label = observation_label
    self._encounter_template = encounter_template
    self._condition_template = condition_template
    self._medication_template = medication_template
    self._procedure_template = procedure_template
    self._observation_template = observation_template
    self._source = source

serialize

serialize(patient: PatientRecord) -> ClinicalTextRecord

Render every encounter/condition/medication/procedure/ observation as one chronological, compact plain-text timeline.

Returns:

Type	Description
ClinicalTextRecord	A ClinicalTextRecord whose patient_ref and
ClinicalTextRecord	deid_status carry over from patient unchanged, ready
ClinicalTextRecord	for the same de-identification/chunking/guardrail pipeline
ClinicalTextRecord	any other clinical text record goes through.

Source code in src/openbtk/pipelines/timeline.py

def serialize(self, patient: PatientRecord) -> ClinicalTextRecord:
    """Render every encounter/condition/medication/procedure/
    observation as one chronological, compact plain-text timeline.

    Returns:
        A ``ClinicalTextRecord`` whose ``patient_ref`` and
        ``deid_status`` carry over from ``patient`` unchanged, ready
        for the same de-identification/chunking/guardrail pipeline
        any other clinical text record goes through.
    """
    rows: list[tuple[datetime | None, str, str]] = []
    for encounter in patient.encounters:
        rows.append(
            (
                encounter.start,
                self._encounter_label,
                self._encounter_template(encounter),
            )
        )
    for condition in patient.conditions:
        rows.append(
            (
                condition.timestamp,
                self._condition_label,
                self._condition_template(condition),
            )
        )
    for medication in patient.medications:
        rows.append(
            (
                medication.timestamp,
                self._medication_label,
                self._medication_template(medication),
            )
        )
    for procedure in patient.procedures:
        rows.append(
            (
                procedure.timestamp,
                self._procedure_label,
                self._procedure_template(procedure),
            )
        )
    for observation in patient.observations:
        rows.append(
            (
                observation.timestamp,
                self._observation_label,
                self._observation_template(observation),
            )
        )
    rows.sort(key=lambda row: (row[0] is None, row[0] or _EPOCH))
    text = "\n".join(self._render_line(*row) for row in rows)
    return ClinicalTextRecord(
        record_id=f"timeline:{patient.patient_id}",
        source=self._source,
        text=text,
        note_type="EHR Timeline",
        patient_ref=patient.patient_id,
        deid_status=patient.deid_status,
    )

Cohorts

openbtk.data.ehr.cohort.CohortBuilder

CohortBuilder(records: Iterable[PatientRecord])

Compose include/exclude predicates over a stream of PatientRecords -- never materialises the source (ADR-0004): a cohort over a million patients has bounded memory, the same guarantee every other streaming component in this project makes.

Example

from openbtk.core.schemas import CodeSystem from openbtk.data.ehr.schemas import CodedEvent, Demographics patients = [ ... PatientRecord( ... patient_id="pt-1", demographics=Demographics(), ... conditions=[CodedEvent(code="385093006", system=CodeSystem.SNOMED)], ... source_system="fhir-r4", ... ), ... PatientRecord( ... patient_id="pt-2", ... demographics=Demographics(), ... source_system="fhir-r4", ... ), ... ] cohort = CohortBuilder(patients).include(has_condition("385093006")) [p.patient_id for p in cohort] == ["pt-1"] True

Source code in src/openbtk/data/ehr/cohort.py

def __init__(self, records: Iterable[PatientRecord]) -> None:
    self._records = records
    self._includes: list[Predicate] = []
    self._excludes: list[Predicate] = []

exclude

exclude(predicate: Predicate) -> CohortBuilder

Add a predicate that, if satisfied, drops a candidate (logical OR across every exclude() call). Returns self for chaining.

Source code in src/openbtk/data/ehr/cohort.py

def exclude(self, predicate: Predicate) -> CohortBuilder:
    """Add a predicate that, if satisfied, drops a candidate (logical
    OR across every ``exclude()`` call). Returns ``self`` for
    chaining."""
    self._excludes.append(predicate)
    return self

include

include(predicate: Predicate) -> CohortBuilder

Add a predicate every candidate must satisfy (logical AND across every include() call). Returns self for chaining.

Source code in src/openbtk/data/ehr/cohort.py

def include(self, predicate: Predicate) -> CohortBuilder:
    """Add a predicate every candidate must satisfy (logical AND
    across every ``include()`` call). Returns ``self`` for chaining."""
    self._includes.append(predicate)
    return self

openbtk.data.ehr.cohort.has_condition

has_condition(code: str, *, system: CodeSystem | None = None) -> Predicate

A predicate matching patients with a condition of code (any coding system, unless system narrows it).

Source code in src/openbtk/data/ehr/cohort.py

def has_condition(code: str, *, system: CodeSystem | None = None) -> Predicate:
    """A predicate matching patients with a condition of ``code`` (any
    coding system, unless ``system`` narrows it)."""

    def predicate(patient: PatientRecord) -> bool:
        return _matches_coded_event(patient.conditions, code, system)

    return predicate

openbtk.data.ehr.cohort.has_medication

has_medication(code: str, *, system: CodeSystem | None = None) -> Predicate

See :func:has_condition; matches patient.medications.

Source code in src/openbtk/data/ehr/cohort.py

def has_medication(code: str, *, system: CodeSystem | None = None) -> Predicate:
    """See :func:`has_condition`; matches ``patient.medications``."""

    def predicate(patient: PatientRecord) -> bool:
        return _matches_coded_event(patient.medications, code, system)

    return predicate

openbtk.data.ehr.cohort.has_procedure

has_procedure(code: str, *, system: CodeSystem | None = None) -> Predicate

See :func:has_condition; matches patient.procedures.

Source code in src/openbtk/data/ehr/cohort.py

def has_procedure(code: str, *, system: CodeSystem | None = None) -> Predicate:
    """See :func:`has_condition`; matches ``patient.procedures``."""

    def predicate(patient: PatientRecord) -> bool:
        return _matches_coded_event(patient.procedures, code, system)

    return predicate

openbtk.data.ehr.cohort.age_between

age_between(min_age: int, max_age: int, *, as_of: date | None = None) -> Predicate

A predicate matching patients whose age -- computed from Demographics.birth_date as of as_of -- falls in [min_age, max_age] inclusive.

as_of defaults to date.today() evaluated the moment the returned predicate is called, not when this factory is, so a predicate built once and reused across a long-running process stays correct. A patient with no known birth_date never matches (age is genuinely unknown, not assumed to be anything).

Source code in src/openbtk/data/ehr/cohort.py

def age_between(min_age: int, max_age: int, *, as_of: date | None = None) -> Predicate:
    """A predicate matching patients whose age -- computed from
    ``Demographics.birth_date`` as of ``as_of`` -- falls in
    ``[min_age, max_age]`` inclusive.

    ``as_of`` defaults to ``date.today()`` evaluated the moment the
    returned predicate is *called*, not when this factory is, so a
    predicate built once and reused across a long-running process stays
    correct. A patient with no known ``birth_date`` never matches (age is
    genuinely unknown, not assumed to be anything).
    """

    def predicate(patient: PatientRecord) -> bool:
        birth_date = patient.demographics.birth_date
        if birth_date is None:
            return False
        reference = as_of if as_of is not None else datetime.now(tz=UTC).date()
        had_birthday_this_year = (reference.month, reference.day) >= (
            birth_date.month,
            birth_date.day,
        )
        age = reference.year - birth_date.year - (0 if had_birthday_this_year else 1)
        return min_age <= age <= max_age

    return predicate

openbtk.data.ehr.cohort.quasi_identifiers

quasi_identifiers(records: Iterable[PatientRecord], fields: Sequence[str] = ('birth_year', 'gender', 'race', 'ethnicity'), *, as_of: date | None = None, include_id: bool = False) -> Iterator[dict[str, object]]

Yield one row per patient with the attributes a person could be linked on.

The bridge from a cohort to openbtk.deid.kanonymity.k_anonymity_report and anonymise_to_k. Streaming: one row is built per record and nothing is retained.

Parameters:

Name	Type	Description	Default
records	Iterable[PatientRecord]	The cohort.	required
fields	Sequence[str]	Which attributes to read: birth_year, age (whole years as of as_of), gender, race, ethnicity, admission_year and admission_month (YYYY-MM), the latter two from the earliest encounter. An attribute a record does not have is None.	('birth_year', 'gender', 'race', 'ethnicity')
as_of	date | None	The date age is computed on. Required when age is asked for, because an age that depends on today's date would make results differ between runs.	None
include_id	bool	Also emit patient_id. It is never a quasi-identifier; it is for keep= when you need to join the exported table back.	False

Raises:

Type	Description
ValueError	On an unknown field, or age without as_of.

Example

from openbtk.data.ehr.schemas import Demographics patient = PatientRecord( ... patient_id="pt-1", ... demographics=Demographics(birth_date=date(1980, 6, 1), gender="female"), ... source_system="fhir-r4", ... ) list(quasi_identifiers([patient], ["birth_year", "gender"])) [{'birth_year': 1980, 'gender': 'female'}]

Source code in src/openbtk/data/ehr/cohort.py

def quasi_identifiers(
    records: Iterable[PatientRecord],
    fields: Sequence[str] = ("birth_year", "gender", "race", "ethnicity"),
    *,
    as_of: date | None = None,
    include_id: bool = False,
) -> Iterator[dict[str, object]]:
    """Yield one row per patient with the attributes a person could be linked on.

    The bridge from a cohort to ``openbtk.deid.kanonymity.k_anonymity_report`` and
    ``anonymise_to_k``. Streaming: one row is built per record and nothing is retained.

    Args:
        records: The cohort.
        fields: Which attributes to read: ``birth_year``, ``age`` (whole years as of
            ``as_of``), ``gender``, ``race``, ``ethnicity``, ``admission_year`` and
            ``admission_month`` (``YYYY-MM``), the latter two from the earliest
            encounter. An attribute a record does not have is ``None``.
        as_of: The date ``age`` is computed on. Required when ``age`` is asked for,
            because an age that depends on today's date would make results differ
            between runs.
        include_id: Also emit ``patient_id``. It is never a quasi-identifier; it is for
            ``keep=`` when you need to join the exported table back.

    Raises:
        ValueError: On an unknown field, or ``age`` without ``as_of``.

    Example:
        >>> from openbtk.data.ehr.schemas import Demographics
        >>> patient = PatientRecord(
        ...     patient_id="pt-1",
        ...     demographics=Demographics(birth_date=date(1980, 6, 1), gender="female"),
        ...     source_system="fhir-r4",
        ... )
        >>> list(quasi_identifiers([patient], ["birth_year", "gender"]))
        [{'birth_year': 1980, 'gender': 'female'}]
    """
    unknown = [f for f in fields if f not in QUASI_IDENTIFIER_FIELDS]
    if unknown:
        raise ValueError(
            f"unknown quasi-identifier(s) {unknown}; choose from "
            f"{list(QUASI_IDENTIFIER_FIELDS)}"
        )
    if "age" in fields and as_of is None:
        raise ValueError("age needs as_of=, so the result does not depend on today")
    for record in records:
        demographics = record.demographics
        starts = [e.start for e in record.encounters if e.start is not None]
        first = min(starts) if starts else None
        values: dict[str, object] = {
            "birth_year": demographics.birth_date.year
            if demographics.birth_date
            else None,
            "age": (
                _age_on(demographics.birth_date, as_of)
                if demographics.birth_date and as_of
                else None
            ),
            "gender": demographics.gender,
            "race": demographics.race,
            "ethnicity": demographics.ethnicity,
            "admission_year": first.year if first else None,
            "admission_month": f"{first.year:04d}-{first.month:02d}" if first else None,
        }
        row = {name: values[name] for name in fields}
        if include_id:
            row["patient_id"] = record.patient_id
        yield row