Skip to main content

setup

Shared assembly of a background-DAG DAGRunContext.

Both the DATASET_PROJECT_LINKED gRPC trigger (Pod handler) and the run_background_task CLI need to build the same DAGRunContext for a background DAG: the file-metadata context provider, the runtime objects injected into every step (datasource, model_resource, schema, datastructure, cache, task_hash, run_id), and the completion reporting that closes out both the Hub task and the cache Run row.

This module is the single source of truth for that assembly so the two entry points cannot drift apart (a drift that has caused repeated ParameterBindError failures when one side gained a runtime param the other did not supply).

Module​

Functions​

build_background_run_context​

def build_background_run_context(    *,    dag: BackgroundDAG,    flow_spec: FlowSpec,    ds_container: DatasourceContainer,    cache: CacheProtocol,    task_hash: str,    hub: BitfountHub,    pod_identifier: str,    project_id: str,    monitor_task_id: str,    ehr_config: EHRConfig | None,    ehr_secrets: RefreshableJWT | None = None,) ‑> DAGRunContext:

Assemble the DAGRunContext for a background DAG run.

Opens the DAG-level Run row (yielding the run_id threaded into every step for inference-row provenance), builds the DataStructure and model resource, and wires completion reporting to both the Hub and the cache Run lifecycle.

Called from inside the background flow body (see flows.dag.executor.build_background_flow), which is what makes a failure here a visible failed flow run rather than a dropped trigger.

Arguments

  • dag: The validated background DAG (its name is used as the run type).
  • flow_spec: The parsed FlowSpec supplying the data-structure config.
  • ds_container: The datasource container providing the datasource and its loaded schema.
  • cache: The background cache instance.
  • task_hash: Pipeline task hash identifying this run.
  • hub: An authenticated Hub instance (model access + completion events).
  • pod_identifier: The pod identity used as senderId in Hub events.
  • project_id: Project ID gating model access (stable across runs). Used only for HubModelResource, NOT for keying completion events.
  • monitor_task_id: The per-run Hub monitor task id (the triggering message's envelope task_id) that TASK_COMPLETE / TASK_ABORT events are attributed to. The Hub's taskStatusCalculator keys status by this id, so it must be the run-specific task id, not the project id.
  • ehr_config: Config to supply to the EHRDataResource. Can be None if tasks do not rely on EHR.
  • ehr_secrets: EHR JWT secrets to supply to the EHRDataResource, so its FHIR client can authenticate and refresh its own token. Not needed for SMARTBackendEHRConfig (which carries its own auth) or for an unauthenticated server under allow_no_ehr_secrets; required otherwise, or every FHIR request the run makes returns HTTP 401.

Returns A ready DAGRunContext.

Note When ehr_config is present but has no usable credentials, EHRDataResource construction raises ValueError (via build_fhir_token_source), and this function degrades to ehr_data_resource=None rather than letting that abort the run. That function's own docstring argues failing at construction "beats discovering it as an HTTP 401 once per patient at query time" — true only if the query then runs unauthenticated. Under a None EHR resource it does not: ehr_query v3 skips the fetch entirely and serves each patient's last stored ehr_data row instead, so there is no 401-per-patient to fall through to. Degrading here trades "abort the whole run" for "this run has no fresh EHR data," which is the behaviour best-effort serving exists to provide. This is distinct from ehr_config=None (no EHR configured at all): that case's resource identity is uncomputable, which _step_resources (flows.dag.hashing) still fails on via EHRNotConfiguredError, not a ValueError, so it is unaffected by this degrade path.

The catch is scoped to ValueError and nothing broader, so a TypeError or other programming fault from EHRDataResource construction still aborts the run. It also re-raises when ehr_config.base_url is falsy: an empty or missing base_url is a structurally unusable config (e.g. a templated config with an unresolved variable, which desert's schema validation does not catch), not a credential-lifecycle event, and D8's degrade case is specifically "identity computable, credentials missing" — not "config broken." EHRDataResource.__init__ also raises ValueError for an unrecognised provider, which this catch cannot distinguish from a credentials failure without duplicating the provider dispatch; the log message below is worded so it does not assert "no usable credentials" as the cause, precisely because that route can also land here.

build_interactive_run_context​

def build_interactive_run_context(    *,    dag: InteractiveDAG,    flow_spec: FlowSpec,    ds_container: DatasourceContainer,    cache: CacheProtocol,    task_hash: str,    hub: BitfountHub,    pod_identifier: str,    project_id: str,    monitor_task_id: str,    ehr_config: EHRConfig | None,    worker_mailbox: _LifecycleMailbox | None = None,    run_id: str | None = None,) ‑> DAGRunContext | None:

Assemble the DAGRunContext for an interactive DAG run.

Mirrors build_background_run_context but additionally wires the BackgroundResultsContext provider (so interactive steps can resolve cross-phase BackgroundRef inputs from cache) and — for worker_only flows triggered over gRPC — a LifecycleNotifier with the initiator's mailbox, so the terminal TASK_COMPLETE / TASK_ABORT are sent back to the JOB_REQUEST sender alongside the cache Run-row bookkeeping. The pre-run signals (JOB_ACCEPT / PREPARING_DATA / CONFIGURING_TASK) are sent by the gRPC handler before this run is scheduled, not here.

Arguments

  • dag: The validated interactive DAG (carries the background steps that back the cross-phase results provider).
  • flow_spec: The parsed FlowSpec supplying the data-structure config.
  • ds_container: The datasource container providing the datasource and its loaded schema.
  • cache: The background cache instance (must already hold the background phase's inference rows).
  • task_hash: Pipeline task hash identifying this run.
  • hub: An authenticated Hub instance (model access + completion events).
  • pod_identifier: The pod identity used as senderId in Hub events.
  • project_id: Project ID gating model access (used for HubModelResource).
  • monitor_task_id: The per-run Hub monitor task id that TASK_COMPLETE / TASK_ABORT events are attributed to.
  • ehr_config: The pod's EHR configuration, or None when it has none. Not used to build a resource here — the interactive phase runs no EHR step today — but required to re-derive the task_hash of a background EHR step, whose hash folds in the EHR's identity. Pass pod.ehr_config, the same value the background phase was given, or the cross-phase read misses. Deliberately has no default: it is required on both run-context builders, so a caller that forgets it fails at the call rather than silently re-deriving a background EHR step's hash without the EHR.
  • worker_mailbox: When the run was triggered by a gRPC JOB_REQUEST, the mailbox addressed to the initiator. Enables message-service lifecycle signalling for worker_only flows. None for CLI runs (no message service). Typed as the _LifecycleMailbox base because that is all this function needs — it only ever forwards the mailbox to LifecycleNotifier — and the gRPC handler's mailbox builder is declared against the base too.
  • run_id: A pre-opened Run row id. Pass this when the caller (the gRPC handler) has already called start_interactive_run — the caller then owns closing the row out on setup failure. When None (the CLI path) this function opens the row itself and marks it failed on an assembly error.

Returns A ready DAGRunContext, or None when a non-stale interactive run is already active for this (task_hash, dag.name) (only possible when this function opens the row itself, i.e. run_id was not supplied).

datasource_type_identity​

def datasource_type_identity(datasource: Any) ‑> str:

Return an identity for datasource's type, for per-file partitions.

Steps whose table is keyed per file key their partition on this instead of on the datasource-name leaf hash, so a file present in several same-type datasources on one pod is processed once (see flows/dag/hashing). The class name is the discriminator because it is what decides how bytes are decoded: a DICOMOphthalmologySource and a HeidelbergE2ESource reading identical bytes do not produce identical rows.

Because the identity is the class name, renaming a datasource class re-partitions every entity-keyed table it feeds — every file re-inferred for an edit that changes no bytes. test_setup.py's test_real_datasource_class_names_are_pinned pins the real classes' names, so a rename fails there and the cache cost is visible when the decision is made.

Known limitation, deliberately accepted: two datasources of the same class can still differ in content-affecting constructor options (e.g. DICOMOphthalmologySource's required_fields or dicom_ophthalmology_csv_columns), and this does not distinguish them. Those options cannot be enumerated reliably — every datasource __init__ takes **kwargs and forwards it up a deep hierarchy — so a best-effort allowlist would give false confidence. What guards the common case instead is the per-row content hash (model_inferences.source_file_hash), which catches the file changing underneath a cached row; a same-type/different-options pair remains a known gap.

Arguments

  • datasource: The run's datasource.

Returns The datasource's class name, or "" when it has none (nothing to key on, so callers keep the leaf hash).

datastructure_identity​

def datastructure_identity(ds_config: DataStructureConfig) ‑> str:

Return a stable identity for the parts of ds_config a model sees.

data_structure is task-level and reaches a step as the datastructure runtime parameter, not as step config, so nothing else folds it into a step's task_hash — yet transform.image, select and assign all change the tensor model_inference feeds the model. See flows/dag/hashing.

Derived from exactly the fields DataStructure.create_datastructure is given below, minus _NON_DETERMINANT_DATASTRUCTURE_FIELDS and minus schema (which is datasource-derived, and whose identity the datasource leaf hash already carries). tests/bitfount/flows/dag/test_setup.py pins that correspondence, so a field added to the call cannot silently escape the hash.

Arguments

  • ds_config: The flow's data_structure block.

Returns A canonical JSON string. Stable across processes: asdict yields plain containers and sort_keys=True fixes the ordering, so there is no set-iteration or dict-insertion order to vary.

Raises

  • TypeError: If a field does not canonicalise to JSON — deliberately not swallowed, since falling back to repr could embed a memory address and give a different digest every process.

run_background_task_from_yaml​

def run_background_task_from_yaml(    *,    path_to_task_yaml: str | PathLike[str],    pod_config: str | PathLike[str],    datasource_name: str,    project_id: str | None = None,) ‑> None:

Run a background DAG from a FlowSpec YAML file (CLI entry point).

See bitfount.scripts.run_background_task for the user-facing docs.

run_interactive_task_from_yaml​

def run_interactive_task_from_yaml(    *,    path_to_task_yaml: str | PathLike[str],    pod_config: str | PathLike[str],    datasource_name: str,    project_id: str | None = None,) ‑> None:

Run an interactive DAG from a FlowSpec YAML file (CLI entry point).

See bitfount.scripts.run_interactive_task for the user-facing docs.

stamp_dag_partitions​

def stamp_dag_partitions(    dag: BackgroundDAG | InteractiveDAG,    *,    task_hash: str,    flow_spec: FlowSpec,    ds_container: DatasourceContainer,    ehr_config: EHRConfig | None,) ‑> dict[str, str]:

Stamp per-step cache partition keys onto dag from a run's own inputs.

The single place the arguments to stamp_step_task_hashes are assembled. Both run-context builders and the v9 end-to-end helpers go through here, because a caller that assembles a subset of them still produces valid-looking digests — just ones naming partitions the run never wrote. datastructure_identity and datasource_type_identity both move a model_inference digest, so dropping either is silent rather than loud.

Arguments

  • dag: The parsed DAG to stamp, mutated in place.
  • task_hash: The datasource leaf hash.
  • flow_spec: The parsed spec, for its data_structure block.
  • ds_container: The run's datasource container, for its datasource type.
  • ehr_config: The pod's EHR configuration, or None.

Returns Mapping of step name -> assigned hash, for logging and assertions.

start_interactive_run​

def start_interactive_run(    cache: CacheProtocol, task_hash: str, *, dag_name: str,) ‑> str | None:

Open the interactive DAG's cache Run row and return its run_id.

Returns None when a non-stale interactive run is already active for this (task_hash, dag_name) — the caller should skip rather than duplicate.

Exposed so the gRPC handler can open the row and send JOB_ACCEPT together (accepting a job and recording it as started is one fact), then hand the resulting run_id to build_interactive_run_context.