iris.orchestration package

Submodules

iris.orchestration.environment module

class iris.orchestration.environment.Environment(*, call_trace_initialiser: Callable[[Dict[str, Algorithm], List[PipelineNode]], PipelineCallTraceStorage], pipeline_output_builder: Callable[[PipelineCallTraceStorage], Any], error_manager: Callable[[PipelineCallTraceStorage, Exception], None], disabled_qa: List[type] = [])

Bases: ImmutableModel

Data holder for the pipeline environment properties.

call_trace_initialiser is responsible for initialising the PipelineCallTraceStorage instance in the pipeline.

pipeline_output_builder is responsible for building the pipeline output from the call_trace, which kept all intermediary results so far.

error_manager is responsible for the pipeline’s behaviour in case of an exception

disabled_qa stores a list of Algorithm and/or Callbacks types to be disabled.

call_trace_initialiser: Callable[[Dict[str, Algorithm], List[PipelineNode]], PipelineCallTraceStorage]

disabled_qa: List[type]

error_manager: Callable[[PipelineCallTraceStorage, Exception], None]

pipeline_output_builder: Callable[[PipelineCallTraceStorage], Any]

iris.orchestration.error_managers module

iris.orchestration.error_managers.raise_error_manager(call_trace: PipelineCallTraceStorage, exception: Exception) → None

Error manager for the Orb.

Parameters:

• call_trace (PipelineCallTraceStorage) – Pipeline call results storage.

• exception (Exception) – Exception raised during the pipeline call.

Raises:

Exception – Reraise the exception parameter.

iris.orchestration.error_managers.store_error_manager(call_trace: PipelineCallTraceStorage, exception: Exception) → None

Error manager for debugging.

Parameters:

• call_trace (PipelineCallTraceStorage) – Pipeline call results storage.

• exception (Exception) – Exception raised during the pipeline call.

iris.orchestration.output_builders module

iris.orchestration.output_builders.build_debugging_output(call_trace: PipelineCallTraceStorage) → Dict[str, Any]

Build the output for debugging purposes.

Parameters:

call_trace (PipelineCallTraceStorage) – Pipeline call results storage.

Returns:

Returns data to be stored in MongoDB.

Return type:

Dict[str, Any]

iris.orchestration.output_builders.build_orb_output(call_trace: PipelineCallTraceStorage) → Dict[str, Any]

Build the output for the Orb.

Parameters:

call_trace (PipelineCallTraceStorage) – Pipeline call results storage.

Returns:

{ “iris_template”: (Optional[Dict]) the iris template dict if the pipeline succeeded. “error”: (Optional[Dict]) the error dict if the pipeline returned an error. “metadata”: (Dict) the metadata dict. }.

Return type:

Dict[str, Any]

iris.orchestration.output_builders.build_simple_debugging_output(call_trace: PipelineCallTraceStorage) → Dict[str, Any]

Build the simplest output for debugging purposes.

Parameters:

call_trace (PipelineCallTraceStorage) – Pipeline call results storage.

Returns:

Returns data to be stored in MongoDB.

Return type:

Dict[str, Any]

iris.orchestration.output_builders.build_simple_orb_output(call_trace: PipelineCallTraceStorage) → Dict[str, Any]

Build the output for the Orb.

Parameters:

call_trace (PipelineCallTraceStorage) – Pipeline call results storage.

Returns:

{ “iris_template”: (Optional[IrisTemplate]) the iris template object if the pipeline succeeded, “error”: (Optional[Dict]) the error dict if the pipeline returned an error, “metadata”: (Dict) the metadata dict, }.

Return type:

Dict[str, Any]

iris.orchestration.pipeline_dataclasses module

class iris.orchestration.pipeline_dataclasses.PipelineClass(*, class_name: str, params: Dict[str, int | float | str | PipelineClass | Tuple[int, float, str, PipelineClass] | List[int | float | str | PipelineClass]])

Bases: ImmutableModel

Data holder for the reference to any class: Algorithm, Callback, etc.

class_name: str

params: Dict[str, int | float | str | PipelineClass | Tuple[int, float, str, PipelineClass] | List[int | float | str | PipelineClass]]

class iris.orchestration.pipeline_dataclasses.PipelineInput(*, name: str, index: int | None = None, source_node: str | List[str | PipelineValue])

Bases: PipelineValue

Data holder for the reference to an input node.

source_node: str | List[str | PipelineValue]

class iris.orchestration.pipeline_dataclasses.PipelineMetadata(*, pipeline_name: str, iris_version: str)

Bases: ImmutableModel

Data holder for input config’s metadata.

iris_version: str

pipeline_name: str

class iris.orchestration.pipeline_dataclasses.PipelineNode(*, name: str, algorithm: PipelineClass, inputs: List[PipelineInput], callbacks: List[PipelineClass] | None = None, seed: str | None = None)

Bases: ImmutableModel

Data holder for one node in a declared pipeline.

algorithm: PipelineClass

callbacks: List[PipelineClass] | None

inputs: List[PipelineInput]

name: str

seed: str | None

class iris.orchestration.pipeline_dataclasses.PipelineValue(*, name: str, index: int | None = None)

Bases: ImmutableModel

Data holder for pipeline value that flows through the system with optional index to specify value this holder refers to if value is of Iterable type.

index: int | None

name: str

iris.orchestration.validators module

iris.orchestration.validators.pipeline_config_duplicate_node_name_check(cls: type, v: List[Any], field: ModelField) → List[Any]

Check if all pipeline nodes have distinct names.

Parameters:

• cls (type) – Class type.

• v (List[Any]) – Value to check.

• field (fields.ModelField) – Field descriptor.

Raises:

IRISPipelineError – Raised if pipeline nodes aren’t unique.

Returns:

v sent for further processing.

Return type:

List[Any]

iris.orchestration.validators.pipeline_metadata_version_check(cls: type, v: str, field: ModelField) → str

Check if the version provided in the input config matches the current iris.__version__.

Module contents