Skip to content

value_refs

Attributes

Classes

StepValueAddress

Bases: BaseModel

Small model to describe the address of a value of a step, within a Pipeline/PipelineStructure.

Source code in kiara/models/module/pipeline/value_refs.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
class StepValueAddress(BaseModel):
    """Small model to describe the address of a value of a step, within a Pipeline/PipelineStructure."""

    class Config:
        extra = Extra.forbid

    step_id: str = Field(description="The id of a step within a pipeline.")
    value_name: str = Field(
        description="The name of the value (output name or pipeline input name)."
    )
    sub_value: Union[Dict[str, Any], None] = Field(
        default=None,
        description="A reference to a subitem of a value (e.g. column, list item)",
    )

    @property
    def alias(self):
        """An alias string for this address (in the form ``[step_id].[value_name]``)."""
        return generate_step_alias(self.step_id, self.value_name)

    def __eq__(self, other):

        if not isinstance(other, StepValueAddress):
            return False

        return (self.step_id, self.value_name, self.sub_value) == (
            other.step_id,
            other.value_name,
            other.sub_value,
        )

    def __hash__(self):

        return hash((self.step_id, self.value_name, self.sub_value))

    def __repr__(self):

        if self.sub_value:
            sub_value = f" sub_value={self.sub_value}"
        else:
            sub_value = ""
        return f"{self.__class__.__name__}(step_id={self.step_id}, value_name={self.value_name}{sub_value})"

    def __str__(self):
        return self.__repr__()

Attributes

step_id: str = Field(description='The id of a step within a pipeline.') class-attribute
value_name: str = Field(description='The name of the value (output name or pipeline input name).') class-attribute
sub_value: Union[Dict[str, Any], None] = Field(default=None, description='A reference to a subitem of a value (e.g. column, list item)') class-attribute
alias property

An alias string for this address (in the form [step_id].[value_name]).

Classes

Config
Source code in kiara/models/module/pipeline/value_refs.py 
24
25
class Config:
    extra = Extra.forbid
Attributes
extra = Extra.forbid class-attribute

ValueRef

Bases: BaseModel

An object that holds information about the location of a value within a pipeline (or other structure).

Basically, a ValueRef helps the containing object where in its structure the value belongs (for example so it can update dependent other values). A ValueRef object (obviously) does not contain the value itself.

There are four different ValueRef type that are relevant for pipelines:

  • [kiara.pipeline.values.StepInputRef][]: an input to a step
  • [kiara.pipeline.values.StepOutputRef][]: an output of a step
  • [kiara.pipeline.values.PipelineInputRef][]: an input to a pipeline
  • [kiara.pipeline.values.PipelineOutputRef][]: an output for a pipeline

Several ValueRef objects can target the same value, for example a step output and a connected step input would reference the same Value (in most cases)..

Source code in kiara/models/module/pipeline/value_refs.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class ValueRef(BaseModel):
    """An object that holds information about the location of a value within a pipeline (or other structure).

    Basically, a `ValueRef` helps the containing object where in its structure the value belongs (for example so
    it can update dependent other values). A `ValueRef` object (obviously) does not contain the value itself.

    There are four different ValueRef type that are relevant for pipelines:

    - [kiara.pipeline.values.StepInputRef][]: an input to a step
    - [kiara.pipeline.values.StepOutputRef][]: an output of a step
    - [kiara.pipeline.values.PipelineInputRef][]: an input to a pipeline
    - [kiara.pipeline.values.PipelineOutputRef][]: an output for a pipeline

    Several `ValueRef` objects can target the same value, for example a step output and a connected step input would
    reference the same `Value` (in most cases)..
    """

    class Config:
        allow_mutation = True
        extra = Extra.forbid

    _id: uuid.UUID = PrivateAttr(default_factory=uuid.uuid4)
    value_name: str
    value_schema: ValueSchema

    def __eq__(self, other):

        if not isinstance(other, self.__class__):
            return False

        return self._id == other._id

    def __hash__(self):
        return hash(self._id)

    def __repr__(self):
        step_id = ""
        if hasattr(self, "step_id"):
            step_id = f" step_id='{self.step_id}'"
        return f"{self.__class__.__name__}(value_name='{self.value_name}' {step_id})"

    def __str__(self):
        name = camel_case_to_snake_case(self.__class__.__name__[0:-5], repl=" ")
        return f"{name}: {self.value_name} ({self.value_schema.type})"

Attributes

value_name: str class-attribute
value_schema: ValueSchema class-attribute

Classes

Config
Source code in kiara/models/module/pipeline/value_refs.py 
85
86
87
class Config:
    allow_mutation = True
    extra = Extra.forbid
Attributes
allow_mutation = True class-attribute
extra = Extra.forbid class-attribute

StepInputRef

Bases: ValueRef

An input to a step.

This object can either have a 'connected_outputs' set, or a 'connected_pipeline_input', not both.

Source code in kiara/models/module/pipeline/value_refs.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
class StepInputRef(ValueRef):
    """An input to a step.

    This object can either have a 'connected_outputs' set, or a 'connected_pipeline_input', not both.
    """

    step_id: str = Field(description="The step id.")
    connected_outputs: Union[List[StepValueAddress], None] = Field(
        default=None,
        description="A potential connected list of one or several module outputs.",
    )
    connected_pipeline_input: Union[str, None] = Field(
        default=None, description="A potential pipeline input."
    )
    is_constant: bool = Field(
        description="Whether this input is a constant and can't be changed by the user."
    )

    @root_validator(pre=True)
    def ensure_single_connected_item(cls, values):

        if values.get("connected_outputs", None) and values.get(
            "connected_pipeline_input"
        ):
            raise ValueError("Multiple connected items, only one allowed.")

        return values

    @property
    def alias(self) -> str:
        return generate_step_alias(self.step_id, self.value_name)

    @property
    def address(self) -> StepValueAddress:
        return StepValueAddress(step_id=self.step_id, value_name=self.value_name)

    def __str__(self):
        name = camel_case_to_snake_case(self.__class__.__name__[0:-5], repl=" ")
        return f"{name}: {self.step_id}.{self.value_name} ({self.value_schema.type})"

Attributes

step_id: str = Field(description='The step id.') class-attribute
connected_outputs: Union[List[StepValueAddress], None] = Field(default=None, description='A potential connected list of one or several module outputs.') class-attribute
connected_pipeline_input: Union[str, None] = Field(default=None, description='A potential pipeline input.') class-attribute
is_constant: bool = Field(description="Whether this input is a constant and can't be changed by the user.") class-attribute
alias: str property
address: StepValueAddress property

Functions

ensure_single_connected_item(values)
Source code in kiara/models/module/pipeline/value_refs.py 
132
133
134
135
136
137
138
139
140
@root_validator(pre=True)
def ensure_single_connected_item(cls, values):

    if values.get("connected_outputs", None) and values.get(
        "connected_pipeline_input"
    ):
        raise ValueError("Multiple connected items, only one allowed.")

    return values

StepOutputRef

Bases: ValueRef

An output to a step.

Source code in kiara/models/module/pipeline/value_refs.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
class StepOutputRef(ValueRef):
    """An output to a step."""

    class Config:
        allow_mutation = True

    step_id: str = Field(description="The step id.")
    pipeline_output: Union[str, None] = Field(
        description="The connected pipeline output."
    )
    connected_inputs: List[StepValueAddress] = Field(
        description="The step inputs that are connected to this step output",
        default_factory=list,
    )

    @property
    def alias(self) -> str:
        return generate_step_alias(self.step_id, self.value_name)

    @property
    def address(self) -> StepValueAddress:
        return StepValueAddress(step_id=self.step_id, value_name=self.value_name)

    def __str__(self):
        name = camel_case_to_snake_case(self.__class__.__name__[0:-5], repl=" ")
        return f"{name}: {self.step_id}.{self.value_name} ({self.value_schema.type})"

Attributes

step_id: str = Field(description='The step id.') class-attribute
pipeline_output: Union[str, None] = Field(description='The connected pipeline output.') class-attribute
connected_inputs: List[StepValueAddress] = Field(description='The step inputs that are connected to this step output', default_factory=list) class-attribute
alias: str property
address: StepValueAddress property

Classes

Config
Source code in kiara/models/module/pipeline/value_refs.py 
158
159
class Config:
    allow_mutation = True
Attributes
allow_mutation = True class-attribute

PipelineInputRef

Bases: ValueRef

An input to a pipeline.

Source code in kiara/models/module/pipeline/value_refs.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
class PipelineInputRef(ValueRef):
    """An input to a pipeline."""

    connected_inputs: List[StepValueAddress] = Field(
        description="The step inputs that are connected to this pipeline input",
        default_factory=list,
    )
    is_constant: bool = Field(
        description="Whether this input is a constant and can't be changed by the user."
    )

    @property
    def alias(self) -> str:
        return generate_step_alias(PIPELINE_PARENT_MARKER, self.value_name)

Attributes

connected_inputs: List[StepValueAddress] = Field(description='The step inputs that are connected to this pipeline input', default_factory=list) class-attribute
is_constant: bool = Field(description="Whether this input is a constant and can't be changed by the user.") class-attribute
alias: str property

PipelineOutputRef

Bases: ValueRef

An output to a pipeline.

Source code in kiara/models/module/pipeline/value_refs.py 
199
200
201
202
203
204
205
206
class PipelineOutputRef(ValueRef):
    """An output to a pipeline."""

    connected_output: StepValueAddress = Field(description="Connected step outputs.")

    @property
    def alias(self) -> str:
        return generate_step_alias(PIPELINE_PARENT_MARKER, self.value_name)

Attributes

connected_output: StepValueAddress = Field(description='Connected step outputs.') class-attribute
alias: str property

Functions

generate_step_alias(step_id: str, value_name)

Source code in kiara/models/module/pipeline/value_refs.py 
17
18
def generate_step_alias(step_id: str, value_name):
    return f"{step_id}.{value_name}"