task

Prefect task for the patient enrichment step.

Reads each configured source file from the enrichment directory and persists its rows as observation-shaped facts, so criteria matching can evaluate an ObservationCriterion against data no EHR API returned.

Module

Functions

patient_enrichment_task

def patient_enrichment_task(    cache: CacheProtocol, task_hash: str, config: PatientEnrichmentConfig | None = None,) ‑> PatientEnrichmentResult:

Ingest every configured enrichment source into the cache.

A source whose file is missing, unreadable, or configured outside the enrichment directory is logged and skipped: enrichment is supplementary, so a broken source should leave its criteria evaluating unknown rather than failing the whole DAG.

A source whose file hashes to what was last ingested is skipped without being re-parsed. A source that has changed has its entire previous row set replaced by the newly parsed rows (replace_source_rows), not merged into them — a merge-only write would leave rows behind for any patient or key dropped from the new file, serving stale data to criteria matching forever.

That replace only happens for a file records_from_csv could actually read. A file it could not interpret — an unparseable CSV, or one missing a required column — raises EnrichmentSourceUnreadableError, which is caught here without calling replace_source_rows at all: nothing is known about the source's current data, so its previously stored rows are left exactly as they were, and no hash is recorded, so the next run retries the read instead of treating the bad state as current.

Arguments

• cache: Cache backend to write the patient_enrichment rows to.
• task_hash: Hash identifying the current task run; written into every row so re-runs upsert rather than duplicating.
• config: The step's config. None, or a config with no sources, stores nothing.

Returns PatientEnrichmentResult with the number of rows persisted by this invocation and a cache accessor scoped to this task_hash.

---