Skip to main content

v1

Patient-eligibility reduce step v1.

Module​

Submodules​

Functions​

task_fn​

def task_fn(    scan_eligibility: CacheAccessor | None,    patient_level_eligibility: CacheAccessor | None,    cache: CacheProtocol,    task_hash: str,    project_id: str,    parent_task_hashes: dict[str, str] | None = None,    config: PatientEligibilityConfig | None = None,    run_id: str | None = None,) ‑> PatientEligibilityResult:

Combine per-scan + patient-level verdicts into per-patient rows; flip pointer.

The rollup rows are always written. The pointer flip is withheld for an empty run, and for one run when this run would publish zero eligible patients on EHR coverage worse than the currently-published partition's — see _withhold_publish, which also documents the case this does not cover. A flip that happens records that coverage and the partition's EHR staleness span in the pointer's tags.

Arguments

  • scan_eligibility: The upstream scan_eligibility edge. Its value is unused — this step reduces the table through direct store calls, which an accessor's grouped queries cannot express — but declaring it is this step's dependency contract, and the name is the key its partition appears under in parent_task_hashes.
  • patient_level_eligibility: The upstream patient_level_eligibility edge, declared on the same basis.
  • cache: The pod background cache to read verdicts from and write rollups to.
  • task_hash: This step's own partition key — where the rollup rows are written, and what the published pointer names for this table.
  • project_id: The trial (project) ID (a runtime param).
  • parent_task_hashes: Partition key of each upstream step, keyed by this step's input parameter name (bitfount.flows.dag.hashing). Where the two source tables are read from — this step's own task_hash would find nothing in them. None because the executor injects it only for a stamped DAG; an unstamped one then reaches the explicit error below rather than a TypeError about a missing argument.
  • config: Optional step config (trial display name); nothing in it is persisted by this step.
  • run_id: Optional provenance run ID.

Returns PatientEligibilityResult with the number of patient_eligibility rows written and a cache accessor scoped to this task_hash.

Raises

  • ValueError: If project_id is not supplied, or parent_task_hashes is missing either upstream partition — without which this would silently roll up zero patients and leave the pointer on the previous run's data.

Classes​

Config​

class Config(**data: Any):

Config for the patient_eligibility reduce step.

Arguments

  • trial_name: Optional display name for the trial, stored for the UI.

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.

Variables​

  • static model_config
  • static trial_name : str | None

Result​

class Result(**data: Any):

Result of a patient_eligibility reduce task.

records_stored reports how many patient_eligibility rows were persisted; 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.