Skip to main content

v1

Patient enrichment step v1.

Module​

Submodules​

Functions​

task_fn​

def task_fn(    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.

Classes​

Config​

class Config(**data: Any):

Config for the patient enrichment step.

sources being None is the whole of the step's optionality: no sources means nothing is read, nothing is stored, and no criterion can be decided on supplied data. There is no separate enable flag and no empty-list sentinel.

Configuring a source is not a purely local change, and this is the paragraph to read before doing it on a live pod. Two effects reach reported eligibility, and neither is a defect:

A criterion's reported provenance follows the row. An observation criterion reads both Observations and Supplied Observations, so each row's evidence is stamped ehr or supplied according to which column decided that row (supplied when both did — see _merged_provenance). A report that previously said ehr for every observation criterion will say supplied for the rows a source actually supplied. That is the point of the feature, but it does mean stored evidence changes shape for a pod that starts supplying data.

A lateralised supplied criterion gates the scan verdict. A criterion with laterality.side != "either" is scan-grain, because the study eye is resolved per scan, so it is partitioned into scan_eligibility and can make scans ineligible. An unlateralised one is patient-grain and gates the patient verdict via patient_level_eligibility. See build_scan_evidence for why supplied criteria are partitioned by grain rather than provenance.

So: a source's first real ingest can change a pod's served verdicts, in the direction the criteria ask for. Configure it deliberately, not as a trial.

Arguments

  • sources: The files to ingest, or None to ingest nothing.

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

Result​

class Result(**data: Any):

Result of a patient enrichment task.

records_stored reports the rows this invocation persisted. A run that skipped every source as unchanged reports 0 — nothing was written — while the partition it stands for still holds the previously ingested rows.

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.