Post Execution
zenml.post_execution
special
After executing a pipeline, the user needs to be able to fetch it from history and perform certain tasks. The post_execution submodule provides a set of interfaces with which the user can interact with artifacts, the pipeline, steps, and the post-run pipeline object.
artifact
ArtifactView
Post-execution artifact class which can be used to read artifact data that was created during a pipeline execution.
Source code in zenml/post_execution/artifact.py
class ArtifactView:
"""Post-execution artifact class which can be used to read
artifact data that was created during a pipeline execution.
"""
def __init__(
self,
id_: int,
type_: str,
uri: str,
materializer: str,
data_type: str,
metadata_store: "BaseMetadataStore",
parent_step_id: int,
):
"""Initializes a post-execution artifact object.
In most cases `ArtifactView` objects should not be created manually but
retrieved from a `StepView` via the `inputs` or `outputs` properties.
Args:
id_: The artifact id.
type_: The type of this artifact.
uri: Specifies where the artifact data is stored.
materializer: Information needed to restore the materializer
that was used to write this artifact.
data_type: The type of data that was passed to the materializer
when writing that artifact. Will be used as a default type
to read the artifact.
metadata_store: The metadata store which should be used to fetch
additional information related to this pipeline.
parent_step_id: The ID of the parent step.
"""
self._id = id_
self._type = type_
self._uri = uri
self._materializer = materializer
self._data_type = data_type
self._metadata_store = metadata_store
self._parent_step_id = parent_step_id
@property
def id(self) -> int:
"""Returns the artifact id."""
return self._id
@property
def type(self) -> str:
"""Returns the artifact type."""
return self._type
@property
def data_type(self) -> str:
"""Returns the data type of the artifact."""
return self._data_type
@property
def uri(self) -> str:
"""Returns the URI where the artifact data is stored."""
return self._uri
@property
def parent_step_id(self) -> int:
"""Returns the ID of the parent step. This need not be equivalent to
the ID of the producer step."""
return self._parent_step_id
@property
def producer_step(self) -> "StepView":
"""Returns the original StepView that produced the artifact."""
# TODO [ENG-174]: Replace with artifact.id instead of passing self if
# required.
return self._metadata_store.get_producer_step_from_artifact(self)
@property
def is_cached(self) -> bool:
"""Returns True if artifact was cached in a previous run, else False."""
# self._metadata_store.
return self.producer_step.id != self.parent_step_id
def read(
self,
output_data_type: Optional[Type[Any]] = None,
materializer_class: Optional[Type["BaseMaterializer"]] = None,
) -> Any:
"""Materializes the data stored in this artifact.
Args:
output_data_type: The datatype to which the materializer should
read, will be passed to the materializers `handle_input` method.
materializer_class: The class of the materializer that should be
used to read the artifact data. If no materializer class is
given, we use the materializer that was used to write the
artifact during execution of the pipeline.
Returns:
The materialized data.
"""
if not materializer_class:
materializer_class = source_utils.load_source_path_class(
self._materializer
)
if not output_data_type:
output_data_type = source_utils.load_source_path_class(
self._data_type
)
logger.debug(
"Using '%s' to read '%s' (uri: %s).",
materializer_class.__qualname__,
self._type,
self._uri,
)
# TODO [ENG-162]: passing in `self` to initialize the materializer only
# works because materializers only require a `.uri` property at the
# moment.
materializer = materializer_class(self) # type: ignore[arg-type]
return materializer.handle_input(output_data_type)
def __repr__(self) -> str:
"""Returns a string representation of this artifact."""
return (
f"{self.__class__.__qualname__}(id={self._id}, "
f"type='{self._type}', uri='{self._uri}', "
f"materializer='{self._materializer}')"
)
def __eq__(self, other: Any) -> bool:
"""Returns whether the other object is referring to the
same artifact."""
if isinstance(other, ArtifactView):
return self._id == other._id and self._uri == other._uri
return NotImplemented
data_type: str
property
readonly
Returns the data type of the artifact.
id: int
property
readonly
Returns the artifact id.
is_cached: bool
property
readonly
Returns True if artifact was cached in a previous run, else False.
parent_step_id: int
property
readonly
Returns the ID of the parent step. This need not be equivalent to the ID of the producer step.
producer_step: StepView
property
readonly
Returns the original StepView that produced the artifact.
type: str
property
readonly
Returns the artifact type.
uri: str
property
readonly
Returns the URI where the artifact data is stored.
__eq__(self, other)
special
Returns whether the other object is referring to the same artifact.
Source code in zenml/post_execution/artifact.py
def __eq__(self, other: Any) -> bool:
"""Returns whether the other object is referring to the
same artifact."""
if isinstance(other, ArtifactView):
return self._id == other._id and self._uri == other._uri
return NotImplemented
__init__(self, id_, type_, uri, materializer, data_type, metadata_store, parent_step_id)
special
Initializes a post-execution artifact object.
In most cases ArtifactView
objects should not be created manually but
retrieved from a StepView
via the inputs
or outputs
properties.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
id_ |
int |
The artifact id. |
required |
type_ |
str |
The type of this artifact. |
required |
uri |
str |
Specifies where the artifact data is stored. |
required |
materializer |
str |
Information needed to restore the materializer that was used to write this artifact. |
required |
data_type |
str |
The type of data that was passed to the materializer when writing that artifact. Will be used as a default type to read the artifact. |
required |
metadata_store |
BaseMetadataStore |
The metadata store which should be used to fetch additional information related to this pipeline. |
required |
parent_step_id |
int |
The ID of the parent step. |
required |
Source code in zenml/post_execution/artifact.py
def __init__(
self,
id_: int,
type_: str,
uri: str,
materializer: str,
data_type: str,
metadata_store: "BaseMetadataStore",
parent_step_id: int,
):
"""Initializes a post-execution artifact object.
In most cases `ArtifactView` objects should not be created manually but
retrieved from a `StepView` via the `inputs` or `outputs` properties.
Args:
id_: The artifact id.
type_: The type of this artifact.
uri: Specifies where the artifact data is stored.
materializer: Information needed to restore the materializer
that was used to write this artifact.
data_type: The type of data that was passed to the materializer
when writing that artifact. Will be used as a default type
to read the artifact.
metadata_store: The metadata store which should be used to fetch
additional information related to this pipeline.
parent_step_id: The ID of the parent step.
"""
self._id = id_
self._type = type_
self._uri = uri
self._materializer = materializer
self._data_type = data_type
self._metadata_store = metadata_store
self._parent_step_id = parent_step_id
__repr__(self)
special
Returns a string representation of this artifact.
Source code in zenml/post_execution/artifact.py
def __repr__(self) -> str:
"""Returns a string representation of this artifact."""
return (
f"{self.__class__.__qualname__}(id={self._id}, "
f"type='{self._type}', uri='{self._uri}', "
f"materializer='{self._materializer}')"
)
read(self, output_data_type=None, materializer_class=None)
Materializes the data stored in this artifact.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
output_data_type |
Optional[Type[Any]] |
The datatype to which the materializer should
read, will be passed to the materializers |
None |
materializer_class |
Optional[Type[BaseMaterializer]] |
The class of the materializer that should be used to read the artifact data. If no materializer class is given, we use the materializer that was used to write the artifact during execution of the pipeline. |
None |
Returns:
Type | Description |
---|---|
Any |
The materialized data. |
Source code in zenml/post_execution/artifact.py
def read(
self,
output_data_type: Optional[Type[Any]] = None,
materializer_class: Optional[Type["BaseMaterializer"]] = None,
) -> Any:
"""Materializes the data stored in this artifact.
Args:
output_data_type: The datatype to which the materializer should
read, will be passed to the materializers `handle_input` method.
materializer_class: The class of the materializer that should be
used to read the artifact data. If no materializer class is
given, we use the materializer that was used to write the
artifact during execution of the pipeline.
Returns:
The materialized data.
"""
if not materializer_class:
materializer_class = source_utils.load_source_path_class(
self._materializer
)
if not output_data_type:
output_data_type = source_utils.load_source_path_class(
self._data_type
)
logger.debug(
"Using '%s' to read '%s' (uri: %s).",
materializer_class.__qualname__,
self._type,
self._uri,
)
# TODO [ENG-162]: passing in `self` to initialize the materializer only
# works because materializers only require a `.uri` property at the
# moment.
materializer = materializer_class(self) # type: ignore[arg-type]
return materializer.handle_input(output_data_type)
pipeline
PipelineView
Post-execution pipeline class which can be used to query pipeline-related information from the metadata store.
Source code in zenml/post_execution/pipeline.py
class PipelineView:
"""Post-execution pipeline class which can be used to query
pipeline-related information from the metadata store.
"""
def __init__(
self, id_: int, name: str, metadata_store: "BaseMetadataStore"
):
"""Initializes a post-execution pipeline object.
In most cases `PipelineView` objects should not be created manually
but retrieved using the `get_pipelines()` method of a
`zenml.core.repo.Repository` instead.
Args:
id_: The context id of this pipeline.
name: The name of this pipeline.
metadata_store: The metadata store which should be used to fetch
additional information related to this pipeline.
"""
self._id = id_
self._name = name
self._metadata_store = metadata_store
@property
def name(self) -> str:
"""Returns the name of the pipeline."""
return self._name
@property
def runs(self) -> List["PipelineRunView"]:
"""Returns all stored runs of this pipeline.
The runs are returned in chronological order, so the latest
run will be the last element in this list.
"""
# Do not cache runs as new runs might appear during this objects
# lifecycle
runs = self._metadata_store.get_pipeline_runs(self)
return list(runs.values())
def get_run_names(self) -> List[str]:
"""Returns a list of all run names."""
# Do not cache runs as new runs might appear during this objects
# lifecycle
runs = self._metadata_store.get_pipeline_runs(self)
return list(runs.keys())
def get_run(self, name: str) -> "PipelineRunView":
"""Returns a run for the given name.
Args:
name: The name of the run to return.
Raises:
KeyError: If there is no run with the given name.
"""
run = self._metadata_store.get_pipeline_run(self, name)
if not run:
raise KeyError(
f"No run found for name `{name}`. This pipeline "
f"only has runs with the following "
f"names: `{self.get_run_names()}`"
)
return run
def __repr__(self) -> str:
"""Returns a string representation of this pipeline."""
return (
f"{self.__class__.__qualname__}(id={self._id}, "
f"name='{self._name}')"
)
def __eq__(self, other: Any) -> bool:
"""Returns whether the other object is referring to the
same pipeline."""
if isinstance(other, PipelineView):
return (
self._id == other._id
and self._metadata_store.uuid == other._metadata_store.uuid
)
return NotImplemented
name: str
property
readonly
Returns the name of the pipeline.
runs: List[PipelineRunView]
property
readonly
Returns all stored runs of this pipeline.
The runs are returned in chronological order, so the latest run will be the last element in this list.
__eq__(self, other)
special
Returns whether the other object is referring to the same pipeline.
Source code in zenml/post_execution/pipeline.py
def __eq__(self, other: Any) -> bool:
"""Returns whether the other object is referring to the
same pipeline."""
if isinstance(other, PipelineView):
return (
self._id == other._id
and self._metadata_store.uuid == other._metadata_store.uuid
)
return NotImplemented
__init__(self, id_, name, metadata_store)
special
Initializes a post-execution pipeline object.
In most cases PipelineView
objects should not be created manually
but retrieved using the get_pipelines()
method of a
zenml.core.repo.Repository
instead.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
id_ |
int |
The context id of this pipeline. |
required |
name |
str |
The name of this pipeline. |
required |
metadata_store |
BaseMetadataStore |
The metadata store which should be used to fetch additional information related to this pipeline. |
required |
Source code in zenml/post_execution/pipeline.py
def __init__(
self, id_: int, name: str, metadata_store: "BaseMetadataStore"
):
"""Initializes a post-execution pipeline object.
In most cases `PipelineView` objects should not be created manually
but retrieved using the `get_pipelines()` method of a
`zenml.core.repo.Repository` instead.
Args:
id_: The context id of this pipeline.
name: The name of this pipeline.
metadata_store: The metadata store which should be used to fetch
additional information related to this pipeline.
"""
self._id = id_
self._name = name
self._metadata_store = metadata_store
__repr__(self)
special
Returns a string representation of this pipeline.
Source code in zenml/post_execution/pipeline.py
def __repr__(self) -> str:
"""Returns a string representation of this pipeline."""
return (
f"{self.__class__.__qualname__}(id={self._id}, "
f"name='{self._name}')"
)
get_run(self, name)
Returns a run for the given name.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
name |
str |
The name of the run to return. |
required |
Exceptions:
Type | Description |
---|---|
KeyError |
If there is no run with the given name. |
Source code in zenml/post_execution/pipeline.py
def get_run(self, name: str) -> "PipelineRunView":
"""Returns a run for the given name.
Args:
name: The name of the run to return.
Raises:
KeyError: If there is no run with the given name.
"""
run = self._metadata_store.get_pipeline_run(self, name)
if not run:
raise KeyError(
f"No run found for name `{name}`. This pipeline "
f"only has runs with the following "
f"names: `{self.get_run_names()}`"
)
return run
get_run_names(self)
Returns a list of all run names.
Source code in zenml/post_execution/pipeline.py
def get_run_names(self) -> List[str]:
"""Returns a list of all run names."""
# Do not cache runs as new runs might appear during this objects
# lifecycle
runs = self._metadata_store.get_pipeline_runs(self)
return list(runs.keys())
pipeline_run
PipelineRunView
Post-execution pipeline run class which can be used to query steps and artifact information associated with a pipeline execution.
Source code in zenml/post_execution/pipeline_run.py
class PipelineRunView:
"""Post-execution pipeline run class which can be used to query
steps and artifact information associated with a pipeline execution.
"""
def __init__(
self,
id_: int,
name: str,
executions: List[proto.Execution],
metadata_store: "BaseMetadataStore",
):
"""Initializes a post-execution pipeline run object.
In most cases `PipelineRunView` objects should not be created manually
but retrieved from a `PipelineView` object instead.
Args:
id_: The context id of this pipeline run.
name: The name of this pipeline run.
executions: All executions associated with this pipeline run.
metadata_store: The metadata store which should be used to fetch
additional information related to this pipeline run.
"""
self._id = id_
self._name = name
self._metadata_store = metadata_store
self._executions = executions
self._steps: Dict[str, StepView] = OrderedDict()
@property
def name(self) -> str:
"""Returns the name of the pipeline run."""
return self._name
@property
def status(self) -> ExecutionStatus:
"""Returns the current status of the pipeline run."""
step_statuses = (step.status for step in self.steps)
if any(status == ExecutionStatus.FAILED for status in step_statuses):
return ExecutionStatus.FAILED
elif all(
status == ExecutionStatus.COMPLETED
or status == ExecutionStatus.CACHED
for status in step_statuses
):
return ExecutionStatus.COMPLETED
else:
return ExecutionStatus.RUNNING
@property
def steps(self) -> List[StepView]:
"""Returns all steps that were executed as part of this pipeline run."""
self._ensure_steps_fetched()
return list(self._steps.values())
def get_step_names(self) -> List[str]:
"""Returns a list of all step names."""
self._ensure_steps_fetched()
return list(self._steps.keys())
def get_step(self, name: str) -> StepView:
"""Returns a step for the given name.
Args:
name: The name of the step to return.
Raises:
KeyError: If there is no step with the given name.
"""
self._ensure_steps_fetched()
try:
return self._steps[name]
except KeyError:
raise KeyError(
f"No step found for name `{name}`. This pipeline "
f"run only has steps with the following "
f"names: `{self.get_step_names()}`"
)
def _ensure_steps_fetched(self) -> None:
"""Fetches all steps for this pipeline run from the metadata store."""
if self._steps:
# we already fetched the steps, no need to do anything
return
self._steps = self._metadata_store.get_pipeline_run_steps(self)
def __repr__(self) -> str:
"""Returns a string representation of this pipeline run."""
return (
f"{self.__class__.__qualname__}(id={self._id}, "
f"name='{self._name}')"
)
def __eq__(self, other: Any) -> bool:
"""Returns whether the other object is referring to the same
pipeline run."""
if isinstance(other, PipelineRunView):
return (
self._id == other._id
and self._metadata_store.uuid == other._metadata_store.uuid
)
return NotImplemented
name: str
property
readonly
Returns the name of the pipeline run.
status: ExecutionStatus
property
readonly
Returns the current status of the pipeline run.
steps: List[zenml.post_execution.step.StepView]
property
readonly
Returns all steps that were executed as part of this pipeline run.
__eq__(self, other)
special
Returns whether the other object is referring to the same pipeline run.
Source code in zenml/post_execution/pipeline_run.py
def __eq__(self, other: Any) -> bool:
"""Returns whether the other object is referring to the same
pipeline run."""
if isinstance(other, PipelineRunView):
return (
self._id == other._id
and self._metadata_store.uuid == other._metadata_store.uuid
)
return NotImplemented
__init__(self, id_, name, executions, metadata_store)
special
Initializes a post-execution pipeline run object.
In most cases PipelineRunView
objects should not be created manually
but retrieved from a PipelineView
object instead.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
id_ |
int |
The context id of this pipeline run. |
required |
name |
str |
The name of this pipeline run. |
required |
executions |
List[ml_metadata.proto.metadata_store_pb2.Execution] |
All executions associated with this pipeline run. |
required |
metadata_store |
BaseMetadataStore |
The metadata store which should be used to fetch additional information related to this pipeline run. |
required |
Source code in zenml/post_execution/pipeline_run.py
def __init__(
self,
id_: int,
name: str,
executions: List[proto.Execution],
metadata_store: "BaseMetadataStore",
):
"""Initializes a post-execution pipeline run object.
In most cases `PipelineRunView` objects should not be created manually
but retrieved from a `PipelineView` object instead.
Args:
id_: The context id of this pipeline run.
name: The name of this pipeline run.
executions: All executions associated with this pipeline run.
metadata_store: The metadata store which should be used to fetch
additional information related to this pipeline run.
"""
self._id = id_
self._name = name
self._metadata_store = metadata_store
self._executions = executions
self._steps: Dict[str, StepView] = OrderedDict()
__repr__(self)
special
Returns a string representation of this pipeline run.
Source code in zenml/post_execution/pipeline_run.py
def __repr__(self) -> str:
"""Returns a string representation of this pipeline run."""
return (
f"{self.__class__.__qualname__}(id={self._id}, "
f"name='{self._name}')"
)
get_step(self, name)
Returns a step for the given name.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
name |
str |
The name of the step to return. |
required |
Exceptions:
Type | Description |
---|---|
KeyError |
If there is no step with the given name. |
Source code in zenml/post_execution/pipeline_run.py
def get_step(self, name: str) -> StepView:
"""Returns a step for the given name.
Args:
name: The name of the step to return.
Raises:
KeyError: If there is no step with the given name.
"""
self._ensure_steps_fetched()
try:
return self._steps[name]
except KeyError:
raise KeyError(
f"No step found for name `{name}`. This pipeline "
f"run only has steps with the following "
f"names: `{self.get_step_names()}`"
)
get_step_names(self)
Returns a list of all step names.
Source code in zenml/post_execution/pipeline_run.py
def get_step_names(self) -> List[str]:
"""Returns a list of all step names."""
self._ensure_steps_fetched()
return list(self._steps.keys())
step
StepView
Post-execution step class which can be used to query artifact information associated with a pipeline step.
Source code in zenml/post_execution/step.py
class StepView:
"""Post-execution step class which can be used to query
artifact information associated with a pipeline step.
"""
def __init__(
self,
id_: int,
parents_step_ids: List[int],
entrypoint_name: str,
name: str,
parameters: Dict[str, Any],
metadata_store: "BaseMetadataStore",
):
"""Initializes a post-execution step object.
In most cases `StepView` objects should not be created manually
but retrieved from a `PipelineRunView` object instead.
Args:
id_: The execution id of this step.
parents_step_ids: The execution ids of the parents of this step.
entrypoint_name: The name of this step.
name: The name of this step within the pipeline
parameters: Parameters that were used to run this step.
metadata_store: The metadata store which should be used to fetch
additional information related to this step.
"""
self._id = id_
self._parents_step_ids = parents_step_ids
self._entrypoint_name = entrypoint_name
self._name = name
self._parameters = parameters
self._metadata_store = metadata_store
self._inputs: Dict[str, ArtifactView] = {}
self._outputs: Dict[str, ArtifactView] = {}
@property
def id(self) -> int:
"""Returns the step id."""
return self._id
@property
def parents_step_ids(self) -> List[int]:
"""Returns a list of ID's of all parents of this step."""
return self._parents_step_ids
@property
def parent_steps(self) -> List["StepView"]:
"""Returns a list of all parent steps of this step."""
steps = [
self._metadata_store.get_step_by_id(s)
for s in self.parents_step_ids
]
return steps
@property
def entrypoint_name(self) -> str:
"""Returns the step entrypoint_name.
This name is equal to the name argument passed to the @step decorator
or the actual function name if no explicit name was given.
Examples:
# the step entrypoint_name will be "my_step"
@step(name="my_step")
def my_step_function(...)
# the step entrypoint_name will be "my_step_function"
@step
def my_step_function(...)
"""
return self._entrypoint_name
@property
def name(self) -> str:
"""Returns the name as it is defined in the pipeline.
This name is equal to the name given to the step within the pipeline
context
Examples:
@step()
def my_step_function(...)
@pipeline
def my_pipeline_function(step_a)
p = my_pipeline_function(
step_a = my_step_function()
)
The name will be `step_a`
"""
return self._name
@property
def parameters(self) -> Dict[str, Any]:
"""The parameters used to run this step."""
return self._parameters
@property
def status(self) -> ExecutionStatus:
"""Returns the current status of the step."""
return self._metadata_store.get_step_status(self)
@property
def is_cached(self) -> bool:
"""Returns whether the step is cached or not."""
return self.status == ExecutionStatus.CACHED
@property
def inputs(self) -> Dict[str, ArtifactView]:
"""Returns all input artifacts that were used to run this step."""
self._ensure_inputs_outputs_fetched()
return self._inputs
@property
def input(self) -> ArtifactView:
"""Returns the input artifact that was used to run this step.
Raises:
ValueError: If there were zero or multiple inputs to this step.
"""
if len(self.inputs) != 1:
raise ValueError(
"Can't use the `StepView.input` property for steps with zero "
"or multiple inputs, use `StepView.inputs` instead."
)
return next(iter(self.inputs.values()))
@property
def outputs(self) -> Dict[str, ArtifactView]:
"""Returns all output artifacts that were written by this step."""
self._ensure_inputs_outputs_fetched()
return self._outputs
@property
def output(self) -> ArtifactView:
"""Returns the output artifact that was written by this step.
Raises:
ValueError: If there were zero or multiple step outputs.
"""
if len(self.outputs) != 1:
raise ValueError(
"Can't use the `StepView.output` property for steps with zero "
"or multiple outputs, use `StepView.outputs` instead."
)
return next(iter(self.outputs.values()))
def _ensure_inputs_outputs_fetched(self) -> None:
"""Fetches all step inputs and outputs from the metadata store."""
if self._inputs or self._outputs:
# we already fetched inputs/outputs, no need to do anything
return
self._inputs, self._outputs = self._metadata_store.get_step_artifacts(
self
)
def __repr__(self) -> str:
"""Returns a string representation of this step."""
return (
f"{self.__class__.__qualname__}(id={self._id}, "
f"name='{self.name}', entrypoint_name='{self.entrypoint_name}'"
f"parameters={self._parameters})"
)
def __eq__(self, other: Any) -> bool:
"""Returns whether the other object is referring to the same step."""
if isinstance(other, StepView):
return (
self._id == other._id
and self._metadata_store.uuid == other._metadata_store.uuid
)
return NotImplemented
entrypoint_name: str
property
readonly
Returns the step entrypoint_name.
This name is equal to the name argument passed to the @step decorator or the actual function name if no explicit name was given.
Examples:
the step entrypoint_name will be "my_step"
@step(name="my_step") def my_step_function(...)
the step entrypoint_name will be "my_step_function"
@step def my_step_function(...)
id: int
property
readonly
Returns the step id.
input: ArtifactView
property
readonly
Returns the input artifact that was used to run this step.
Exceptions:
Type | Description |
---|---|
ValueError |
If there were zero or multiple inputs to this step. |
inputs: Dict[str, zenml.post_execution.artifact.ArtifactView]
property
readonly
Returns all input artifacts that were used to run this step.
is_cached: bool
property
readonly
Returns whether the step is cached or not.
name: str
property
readonly
Returns the name as it is defined in the pipeline.
This name is equal to the name given to the step within the pipeline context
Examples:
@step() def my_step_function(...)
@pipeline def my_pipeline_function(step_a)
p = my_pipeline_function( step_a = my_step_function() )
The name will be step_a
output: ArtifactView
property
readonly
Returns the output artifact that was written by this step.
Exceptions:
Type | Description |
---|---|
ValueError |
If there were zero or multiple step outputs. |
outputs: Dict[str, zenml.post_execution.artifact.ArtifactView]
property
readonly
Returns all output artifacts that were written by this step.
parameters: Dict[str, Any]
property
readonly
The parameters used to run this step.
parent_steps: List[StepView]
property
readonly
Returns a list of all parent steps of this step.
parents_step_ids: List[int]
property
readonly
Returns a list of ID's of all parents of this step.
status: ExecutionStatus
property
readonly
Returns the current status of the step.
__eq__(self, other)
special
Returns whether the other object is referring to the same step.
Source code in zenml/post_execution/step.py
def __eq__(self, other: Any) -> bool:
"""Returns whether the other object is referring to the same step."""
if isinstance(other, StepView):
return (
self._id == other._id
and self._metadata_store.uuid == other._metadata_store.uuid
)
return NotImplemented
__init__(self, id_, parents_step_ids, entrypoint_name, name, parameters, metadata_store)
special
Initializes a post-execution step object.
In most cases StepView
objects should not be created manually
but retrieved from a PipelineRunView
object instead.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
id_ |
int |
The execution id of this step. |
required |
parents_step_ids |
List[int] |
The execution ids of the parents of this step. |
required |
entrypoint_name |
str |
The name of this step. |
required |
name |
str |
The name of this step within the pipeline |
required |
parameters |
Dict[str, Any] |
Parameters that were used to run this step. |
required |
metadata_store |
BaseMetadataStore |
The metadata store which should be used to fetch additional information related to this step. |
required |
Source code in zenml/post_execution/step.py
def __init__(
self,
id_: int,
parents_step_ids: List[int],
entrypoint_name: str,
name: str,
parameters: Dict[str, Any],
metadata_store: "BaseMetadataStore",
):
"""Initializes a post-execution step object.
In most cases `StepView` objects should not be created manually
but retrieved from a `PipelineRunView` object instead.
Args:
id_: The execution id of this step.
parents_step_ids: The execution ids of the parents of this step.
entrypoint_name: The name of this step.
name: The name of this step within the pipeline
parameters: Parameters that were used to run this step.
metadata_store: The metadata store which should be used to fetch
additional information related to this step.
"""
self._id = id_
self._parents_step_ids = parents_step_ids
self._entrypoint_name = entrypoint_name
self._name = name
self._parameters = parameters
self._metadata_store = metadata_store
self._inputs: Dict[str, ArtifactView] = {}
self._outputs: Dict[str, ArtifactView] = {}
__repr__(self)
special
Returns a string representation of this step.
Source code in zenml/post_execution/step.py
def __repr__(self) -> str:
"""Returns a string representation of this step."""
return (
f"{self.__class__.__qualname__}(id={self._id}, "
f"name='{self.name}', entrypoint_name='{self.entrypoint_name}'"
f"parameters={self._parameters})"
)