Skip to main content

v3

EHR query step v3 — best-effort EHR priming.

Same config as v1 and v2, same ehr_data table, and the same two lookup modes. What v3 changes is the step's relationship to the run: an unreachable EHR degrades the run instead of ending it. See v3.task for the mechanics and EHRRunOutcome for what the result reports.

The version exists because behaviour, not config, is what differs. A config flag would move the ehr_query partition for every template using EHRQueryConfig — _semantic_config hashes model_dump(mode="json") including defaults (flows/dag/hashing.py) — so ehr_trial_v9 would lose its ehr_data cache and re-query its whole lister-derived cohort for a flag it never sets. It stays on v2 and keeps fail-fast; patient_eligibility_v9 moves here.

patient_eligibility_v9 pays that cost itself, and it is worth stating plainly: "version": step.version is folded into the task hash (flows/dag/hashing.py), so moving that template from ehr_query v2 to v3 moves the hash for this step and every descendant of it, abandoning the ehr_data partition the template had accumulated. Nothing migrates it — ehr_data is task_hash-keyed and no backfill exists — so the first run after the upgrade starts against an empty partition with nothing stored to serve. If the EHR is unreachable on exactly that run, best-effort serving has no prior row to fall back on for anybody: every patient's EHR criteria resolve unknown, the cohort publishes UNKNOWN, and allows_empty_partition = True (below) means nothing aborts to say so. The signals that do fire are ehr_coverage: 0.0 with ehr_criteria_present: true in the pointer's tags, and ehr_query's own DEGRADED_TOTAL log line. A run that does reach the EHR repopulates the partition in one pass and the fallback is real from the second run on. ehr_trial_v9 keeps its own partition precisely because it stays on v2.

Module​

Submodules​

Functions​

task_fn​

def task_fn(    ehr_data_resource: EHRDataResource | None,    cache: CacheProtocol,    task_hash: str,    config: EHRQueryConfig | None = None,    datasource: BaseSource | None = None,    filenames: list[str] | None = None,    patient_ids: CacheAccessor | None = None,) ‑> EHRQueryResult:

Prime the EHR cache for the current run, without failing the run.

Arguments

  • ehr_data_resource: Configured EHR resource (NextGen or FHIR R4). None means no EHR reached this pod's DAG setup: the run continues, nothing is queried, and the result reports EHRRunOutcome.NOT_CONFIGURED.
  • cache: Cache backend to write ehr_data rows to.
  • task_hash: Hash identifying the current task run; written into every row so re-runs upsert via session.merge.
  • config: Step config. config.mode selects between filename-driven and patient-ID-cache-driven lookups, config.fetch_appointments controls whether appointment and encounter history is fetched, and config.observation_codes/config.observation_categories fetch coded Observations matching those codes/categories (merged and deduplicated when both are set). If neither is set, every core FHIR observation-category is fetched by default — see EHRQueryConfig.fetch_options. When config itself is None the defaults are used so existing programmatic callers do not have to pass one.
  • datasource: Imaging datasource. Required in "filename" mode.
  • filenames: Imaging file IDs to process. Required in "filename" mode.
  • patient_ids: Cache accessor over the ehr_patient_ids table (typically wired from ehr_patient_lister.cache in the YAML). Required in "patient_id_cache" mode.

Returns The step result, always carrying a CacheAccessor over this run's ehr_data partition — including when no row was stored, since an unwired accessor costs criteria_matching its EHR columns entirely, and EHR criteria then read as scan evidence with provenance unknown rather than as UNKNOWN EHR evidence.

Raises

  • ValueError: If config.mode is not one this step supports, or the mode's required inputs are missing. A misconfigured step is not an EHR outage.

Classes​

Config​

class Config(**data: Any):

Config for the EHR criteria query step.

mode selects how patients are identified:

"filename" The default. Patients are resolved from imaging file metadata (name + DOB columns on the datasource). Requires datasource and filenames on the task call.

"patient_id_cache" Patients are resolved by ID from a cache accessor wired in via the step's inputs: block (typically ehr_patient_lister.cache). datasource and filenames are not consulted. Records are stored without imaging file IDs or per-scan metadata.

fetch_appointments adds three per-patient EHR calls (previous appointments, previous encounters, next appointment). It is off by default because those calls cost a round-trip each and only the appointment-history criteria consume them. When off, the corresponding cache columns are left NULL, which downstream criteria read as UNKNOWN rather than as a failure.

observation_codes/observation_categories fetch coded Observations (labs/measurements) — unlike conditions/procedures, Observation searches require an explicit code or category parameter on at least one major FHIR backend (Epic's Observation.Search rejects a bare patient-only query — error 59108, "either the category or code parameter must be specified"). observation_codes is precise (e.g. LOINC-coded albumin); observation_categories is a coarser net over the core FHIR observation-category values (e.g. "laboratory", "vital-signs") — Epic's own docs note category assignment is "subjective and may differ across organizations" and recommend code search when consistency matters. Both may be set together; results are merged and deduplicated (as two separate searches — at least one backend rejects a query with both code and category set, so they are never combined into one).

Default, when neither is set (both left None): ehr_criteria_query_task fetches every core FHIR observation-category (ALL_OBSERVATION_CATEGORIES — social history, vital signs, imaging, laboratory, procedure, survey, exam, therapy, activity), so a criteria-tree CodeCriterion referencing any observation code has a chance of finding data without the flow author having to enumerate codes/categories up front. Setting observation_codes (with observation_categories left None) opts out of that wide default in favour of a precise, narrower fetch; setting observation_categories explicitly is respected as-is, not widened back to "all" — this includes setting it to [], which deliberately fetches no categories at all (a full opt-out of category search, distinct from leaving it None). Likewise, an explicit observation_codes: [] opts out of code search without re-triggering the wide category default. The distinction is None (unset) vs. any list including [] (an explicit, respected choice) — never emptiness. The corresponding cache column is NULL only if the fetch itself fails or genuinely finds nothing, which downstream criteria read as UNKNOWN.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors​

Variables​

  • static fetch_appointments : bool
  • static mode : Literal['filename', 'patient_id_cache']
  • static model_config
  • static observation_categories : Optional[list[str]]

Methods​

fetch_options​

def fetch_options(self) ‑> EHRFetchOptions:

Resolve this config into the fetch scope every lookup is asked for.

A method rather than a helper in ehr_query/functions.py, so that every version of the step resolves the defaults documented above identically: functions.py cannot import this module (its own package imports functions, so the two would form an import cycle), and duplicating the resolution per version would let the versions drift apart on what an unset field means.

Both observation fields are read with is not None rather than for truthiness, so an explicitly empty list — a deliberate opt-out of that search — is honoured instead of falling through to the wide default.

Returns The EHRFetchOptions to pass to the EHR resource for every patient in the run.

Result​

class Result(**data: Any):

Result of a best-effort EHR criteria query task.

records_stored reports how many EHR rows this run persisted to the cache; cache exposes the CacheAccessor over the partition for downstream DAG steps. The accessor is present even when this run stored nothing — with it unwired, criteria_matching has no EHR columns to tag and reports EHR criteria as scan evidence with provenance unknown.

The remaining fields exist because a run that could not reach the EHR is otherwise indistinguishable from a quiet success: the write guard withholds a failure record rather than overwriting a stored row, so the attempt leaves no trace in ehr_data at all.

lister_truncated is deliberately absent: the truncation it reports belongs to ehr_patient_lister, a separate step, and this step's template wires no lister at all.

Attributes

  • run_outcome: How well the run primed the serving layer. See EHRRunOutcome. Required, with no default: a result that did not say how the run went must not be readable as a healthy prime.
  • skipped_total_failure: Patients whose lookup produced nothing usable this run and whose failure record the write guard withheld in favour of a stored row. Those patients are served from a row as old as the last run that reached the EHR for them, and nothing in the table records this run's attempt. A failure the guard did write is not counted here — it is readable as that row's error.
  • skipped_unidentified: Rows carrying nothing to look a patient up by. No EHR query was made for them, and no row can record that: the reason fires before a bitfount_patient_id — the ehr_data primary key — exists to key one on.
  • skipped_incomplete_name: Rows whose name did not yield both a given and a family name. Reported here for the same reason.
  • skipped_unparseable_dob: Rows whose date of birth could not be parsed. Reported here for the same reason.
  • groups_unsupported: The fetch groups this backend reported as structurally lacking, unioned over the patients queried. Sourced from the in-memory fetch outcomes, since ehr_data.error holds one patient-scope reason and cannot represent per-group state. Sticky across runs — a configuration finding, not an outage.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors​

Variables​

  • static model_config
  • static skipped_incomplete_name : int
  • static skipped_total_failure : int
  • static skipped_unidentified : int
  • static skipped_unparseable_dob : int
  • cache : CacheAccessor | None - Return the CacheAccessor for reading this result in a DAG pipeline.

  • records : pandas.core.frame.DataFrame - Retrieve this result's rows as a DataFrame.

    The DataFrame is fetched from the cache on each access — it is not stored in the result object, allowing lazy access to large datasets without materialising them in memory until needed.

    Returns: A pandas.DataFrame with one row per cached record.

    Raises: RuntimeError: If no accessor is attached (e.g. the result was serialised across a Prefect task boundary). In a DAG pipeline use <step_name>.cache instead.

Static methods​

from_accessor​

def from_accessor(    accessor: CacheAccessor,    *,    records_stored: int | None = None,    run_outcome: EHRRunOutcome = not_configured,    skipped_total_failure: int = 0,    skipped_unidentified: int = 0,    skipped_incomplete_name: int = 0,    skipped_unparseable_dob: int = 0,    groups_unsupported: tuple[EHRUnavailableReason, ...] = (),) ‑> Self:

Build a result carrying accessor and this run's status.

Widens the base classmethod with the run-status fields, so the task still builds its result in one call rather than constructing it and reaching in to attach the accessor.

Arguments

  • accessor: The CacheAccessor backing <step>.cache, scoped to this run's partition. Passed even when the run stored no rows.
  • records_stored: Rows this run persisted. Defaults to the whole partition's count, per the base class.
  • run_outcome: How well the run primed the serving layer. Defaults to NOT_CONFIGURED rather than PRIMED, so the generic cross-phase reconstruction in flows/dag/context.py — which cannot know how the background run went — cannot claim a healthy prime on its behalf.
  • skipped_total_failure: Failure records the write guard withheld.
  • skipped_unidentified: Rows with no usable query key.
  • skipped_incomplete_name: Rows with an unsplittable name.
  • skipped_unparseable_dob: Rows with an unparseable date of birth.
  • groups_unsupported: Fetch groups the backend structurally lacks.

Returns The result, with the accessor attached.

Methods​

model_post_init​

def model_post_init(self: BaseModel, context: Any, /) ‑> None:

Inherited from:

CacheBackedResult.model_post_init :

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Arguments

  • self: The BaseModel instance.
  • context: The context.