Skip to main content

schema

Dataclass schema definitions for v9 flow YAML files.

load_flow_spec / load_flow_spec_from_path are the single way to turn a task YAML into a FlowSpec: they render the {{ var }} template placeholders and then parse, in that order. Use them rather than open-coding the pair, so no caller can drift into parsing an unrendered spec.

Module​

Functions​

is_v9_flow_spec​

def is_v9_flow_spec(raw_config: Mapping[str, Any]) ‑> bool:

Whether a raw config dict is a v9 FlowSpec rather than a v8 config.

Lets callers handed an arbitrary project config recognise a v8 ModellerConfig before attempting to load it as a FlowSpec, where the mismatch would otherwise surface as a bare marshmallow ValidationError.

Any one of the v9-only fields is enough: only federation is required by the FlowSpec schema, so a valid v9 config may omit roles and worker and lean on their defaults. Requiring all three would misread such a config as v8, and callers treat "not v9" as "nothing to do here" rather than as an error. A config that carries one marker but is otherwise malformed is better off failing FlowSpec validation loudly.

The equivalent check on an already-deserialised object is bitfount.flows.dag.parser._assert_flow_spec. It demands all three fields, which it can do because deserialisation fills roles and worker in from their defaults; that guarantee does not hold for a raw dict.

Arguments

  • raw_config: A config dict, as parsed from YAML or received over the wire.

Returns Whether the config carries any of the v9-only top-level fields.

load_flow_spec​

def load_flow_spec(    raw: dict[str, Any], template_params: dict[str, Any] | None = None,) ‑> FlowSpec:

Render raw's {{ var }} placeholders and parse it into a FlowSpec.

Task YAMLs may be templates. The Hub renders them upstream in production; any path that bypasses the Hub (the CLI runners, tests) must render before parsing, since the schema's field validators reject placeholder syntax. A spec with no template: section renders to itself.

Does not mutate raw: replace_templated_variables works on a deep clone whenever it substitutes anything, and parsing only reads.

Arguments

  • raw: The task YAML as loaded by yaml.safe_load.
  • template_params: Optional overrides for the template's own defaults.

Returns The parsed FlowSpec.

load_flow_spec_from_path​

def load_flow_spec_from_path(    path: str | PathLike[str], template_params: dict[str, Any] | None = None,) ‑> FlowSpec:

Read the task YAML at path and load it via load_flow_spec.

Arguments

  • path: Path to the task YAML.
  • template_params: Optional overrides for the template's own defaults.

Returns The parsed FlowSpec.

Raises

  • FileNotFoundError: If path does not exist.

Classes​

FederationConfig​

class FederationConfig(strategy: str):

Federation strategy configuration.

Variables​

  • static strategy : str

FlowIntermediaryConfig​

class FlowIntermediaryConfig(steps: list[StepDefinition] = []):

Intermediary role definition for v9 flows.

We currently do not have any need for an "intermediary", but this role would sit in between the worker and modeller as a possible intermediate aggregation step.

Again, much like the modeller, the intermediary only concerns itself with the interactive phase of a task, so does not need separate background/interactive portions.

Named FlowIntermediaryConfig to avoid any future collision with a similarly-named class elsewhere.

Variables​

FlowModellerConfig​

class FlowModellerConfig(steps: list[StepDefinition] = []):

Modeller role definition for v9 flows.

A modeller only concerns itself with the interactive phase of a task, and so does not need separate background/interactive portions like WorkerConfig.

Named FlowModellerConfig to avoid collision with the v8 ModellerConfig from modeller_schemas.

Variables​

FlowSpec​

class FlowSpec(    template: Optional[dict[str, TemplateVariablesEntryString | TemplateVariablesEntryNumber | TemplateVariablesEntryNestedArray | TemplateVariablesEntryArray | TemplateVariablesEntryObject | TemplateVariablesEntryObjectArray | TemplateVariablesEntryFilePath | TemplateVariablesEntryModelSlug | TemplateVariablesEntrySchemaColumnName | TemplateVariablesEntrySchemaColumnNameArray | TemplateVariablesEntryBool | TemplateVariablesEntryTaskFilters]] = None,    *,    name: str,    version: str,    federation: FederationConfig,    pods: PodsConfig,    roles: RolesConfig = RolesConfig(worker=RoleAssignment(background=None, interactive=None), modeller=None, intermediary=None),    config: dict[str, Any] | None = None,    data_structure: DataStructureConfig = DataStructureConfig(table_config=None, assign=DataStructureAssignConfig(target=None, image_cols=None, image_prefix=None), select=DataStructureSelectConfig(include=None, include_prefix=None, exclude=None), transform=DataStructureTransformConfig(dataset=None, batch=None, image=None, auto_convert_grayscale_images=True), data_split=None, schema_requirements='partial', compatible_datasources=['CSVSource', 'DICOMSource', 'ImageSource', 'InterMineSource', 'NIFTISource', 'DICOMOphthalmologySource', 'HeidelbergSource', 'HeidelbergE2ESource', 'TopconSource'], filter=None),    worker: WorkerConfig = WorkerConfig(background=PhaseConfig(steps=[]), interactive=PhaseConfig(steps=[])),    modeller: FlowModellerConfig | None = None,    intermediary: FlowIntermediaryConfig | None = None,):

Top-level schema for v9 flow YAML files.

Identified by the presence of v9-specific fields such as federation, roles, or worker.

Ancestors​

Variables​

  • static config : dict[str, typing.Any] | None
  • static name : str
  • static version : str

PhaseConfig​

class PhaseConfig(steps: list[StepDefinition] = []):

A named phase (background or interactive) containing an ordered list of steps.

Variables​

PodsConfig​

class PodsConfig(identifiers: list[str]):

Pod identifier configuration.

Variables​

  • static identifiers : list[str]

RoleAssignment​

class RoleAssignment(background: str | None = None, interactive: str | None = None):

Maps named sub-roles (e.g. background, interactive) to phase names.

Variables​

  • static background : str | None
  • static interactive : str | None

RolesConfig​

class RolesConfig(    worker: RoleAssignment = RoleAssignment(background=None, interactive=None),    modeller: str | None = None,    intermediary: str | None = None,):

Role configuration: declares which roles exist and their sub-roles.

Variables​

  • static intermediary : str | None
  • static modeller : str | None

StepDefinition​

class StepDefinition(    name: str,    task: str,    version: int,    parallel: bool = False,    save_to_cache: bool = False,    inputs: dict[str, str] = {},):

A single step in a flow phase.

Variables​

  • static inputs : dict[str, str]
  • static name : str
  • static parallel : bool
  • static save_to_cache : bool
  • static task : str
  • static version : int

WorkerConfig​

class WorkerConfig(    background: PhaseConfig = PhaseConfig(steps=[]),    interactive: PhaseConfig = PhaseConfig(steps=[]),):

Worker role definition: background and interactive phases.

Variables​