Skip to main content

v2

EHR query step v2.

Behaviourally identical to v1 — the config, result and task function are the same objects, re-exported. The version exists to move the step's task_hash, for the reason given in bitfount.steps.criteria_matching.v2.

Bumped separately because ehr_query is not downstream of criteria_matching: nothing carries that step's move into this one's hash. Its table, ehr_data, is keyed (task_hash, bitfount_patient_id) and get_ehr_data_by_task reads it by task_hash with the patient predicate optional, so it has the same set-wise-read exposure the other three do.

Module​

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:

Extract and cache EHR data for the current run.

Arguments

  • ehr_data_resource: Configured EHR resource (NextGen or FHIR R4).
  • 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 (ALL_OBSERVATION_CATEGORIES) is fetched by default — see EHRQueryConfig's docstring. 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.

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 an EHR criteria query task.

records_stored reports how many EHR rows were persisted to the cache; cache exposes the CacheAccessor over them for downstream DAG steps.

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
  • 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,) ‑> Self:

Inherited from:

CacheBackedResult.from_accessor :

Build a result carrying accessor, in one call rather than two.

_accessor is a PrivateAttr, so it cannot be passed to the constructor and every producer would otherwise construct the result and then reach in to attach the accessor. That two-step is the shape this replaces.

Arguments

  • accessor: The CacheAccessor backing <step>.cache, scoped to the partition this result stands for.
  • records_stored: Rows this result stands for. Defaults to accessor.count() — the whole partition — which is what a step that persisted nothing this run reports. A step that wrote rows passes what it wrote; see the field's docstring for why the two are not interchangeable.

Returns An instance of the calling subclass, 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.