Skip to content

kiara.pipeline.structure

PipelineStep pydantic-model

A step within a pipeline-structure, includes information about it's connection(s) and other metadata.

The links that connect to inputs of the module.

module_config: Mapping[str, Any] pydantic-field

The module config.

module_type: str pydantic-field required

The module type.

processing_stage: int pydantic-field

The stage number this step is executed within the pipeline.

required: bool pydantic-field

Whether this step is required within the workflow.

In some cases, when none of the pipeline outputs have a required input that connects to a step, then it is not necessary for this step to have been executed, even if it is placed before a step in the execution hierarchy. This also means that the pipeline inputs that are connected to this step might not be required.

__eq__(self, other) special

Return self==value.

Source code in kiara/pipeline/structure.py 
def __eq__(self, other):

    if not isinstance(other, PipelineStep):
        return False

    return self._id == other._id

    # # TODO: also check whether _kiara obj is equal?
    # eq = (self.step_id, self.parent_id, self.module, self.processing_stage,) == (
    #     other.step_id,
    #     other.parent_id,
    #     other.module,
    #     other.processing_stage,
    # )
    #
    # if not eq:
    #     return False
    #
    # hs = DeepHash(self.input_links)
    # ho = DeepHash(other.input_links)
    #
    # return hs[self.input_links] == ho[other.input_links]

__hash__(self) special

Return hash(self).

Source code in kiara/pipeline/structure.py 
def __hash__(self):

    return hash(self._id)

    # # TODO: also include _kiara obj?
    # # TODO: figure out whether that can be made to work without deephash
    # hs = DeepHash(self.input_links)
    # return hash(
    #     (
    #         self.step_id,
    #         self.parent_id,
    #         self.module,
    #         self.processing_stage,
    #         hs[self.input_links],
    #     )
    # )

__repr__(self) special

Return repr(self).

Source code in kiara/pipeline/structure.py 
def __repr__(self):

    return f"{self.__class__.__name__}(step_id={self.step_id} module_type={self.module_type} processing_stage={self.processing_stage})"

__str__(self) special

Return str(self).

Source code in kiara/pipeline/structure.py 
def __str__(self):
    return f"step: {self.step_id} (module: {self.module_type})"

PipelineStructure

An object that holds one or several steps, and describes the connections between them.