Skip to content

Stack

zenml.stack

Initialization of the ZenML Stack.

The stack is essentially all the configuration for the infrastructure of your MLOps platform.

A stack is made up of multiple components. Some examples are:

  • An Artifact Store
  • An Orchestrator
  • A Step Operator (Optional)
  • A Container Registry (Optional)

Attributes

__all__ = ['Flavor', 'Stack', 'StackComponent', 'StackValidator', 'StackComponentConfig'] module-attribute

Classes

Flavor

Class for ZenML Flavors.

Attributes
config_class: Type[StackComponentConfig] abstractmethod property

Returns StackComponentConfig config class.

Returns:

Type Description
Type[StackComponentConfig]

The config class.
config_schema: Dict[str, Any] property

The config schema for a flavor.

Returns:

Type Description
Dict[str, Any]

The config schema.
display_name: Optional[str] property

The display name of the flavor.

By default, converts the technical name to a human-readable format. For example, "vm_kubernetes" becomes "VM Kubernetes". Flavors can override this to provide custom display names.

Returns:

Type Description
Optional[str]

The display name of the flavor.
docs_url: Optional[str] property

A url to point at docs explaining this flavor.

Returns:

Type Description
Optional[str]

A flavor docs url.
implementation_class: Type[StackComponent] abstractmethod property

Implementation class for this flavor.

Returns:

Type Description
Type[StackComponent]

The implementation class for this flavor.
logo_url: Optional[str] property

A url to represent the flavor in the dashboard.

Returns:

Type Description
Optional[str]

The flavor logo.
name: str abstractmethod property

The flavor name.

Returns:

Type Description
str

The flavor name.
sdk_docs_url: Optional[str] property

A url to point at SDK docs explaining this flavor.

Returns:

Type Description
Optional[str]

A flavor SDK docs url.
service_connector_requirements: Optional[ServiceConnectorRequirements] property

Service connector resource requirements for service connectors.

Specifies resource requirements that are used to filter the available service connector types that are compatible with this flavor.

Returns:

Type Description
Optional[ServiceConnectorRequirements]

Requirements for compatible service connectors, if a service
Optional[ServiceConnectorRequirements]

connector is required for this flavor.
type: StackComponentType abstractmethod property

The stack component type.

Returns:

Type Description
StackComponentType

The stack component type.
Functions
from_model(flavor_model: FlavorResponse) -> Flavor classmethod

Loads a flavor from a model.

Parameters:

Name Type Description Default
flavor_model FlavorResponse

The model to load from.

 required

Raises:

Type Description
CustomFlavorImportError

If the custom flavor can't be imported.
ImportError

If the flavor can't be imported.

Returns:

Type Description
Flavor

The loaded flavor.
Source code in src/zenml/stack/flavor.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
@classmethod
def from_model(cls, flavor_model: FlavorResponse) -> "Flavor":
    """Loads a flavor from a model.

    Args:
        flavor_model: The model to load from.

    Raises:
        CustomFlavorImportError: If the custom flavor can't be imported.
        ImportError: If the flavor can't be imported.

    Returns:
        The loaded flavor.
    """
    try:
        flavor = source_utils.load(flavor_model.source)()
    except (ModuleNotFoundError, ImportError, NotImplementedError) as err:
        if flavor_model.is_custom:
            flavor_module, _ = flavor_model.source.rsplit(".", maxsplit=1)
            expected_file_path = os.path.join(
                source_utils.get_source_root(),
                flavor_module.replace(".", os.path.sep),
            )
            raise CustomFlavorImportError(
                f"Couldn't import custom flavor {flavor_model.name}: "
                f"{err}. Make sure the custom flavor class "
                f"`{flavor_model.source}` is importable. If it is part of "
                "a library, make sure it is installed. If "
                "it is a local code file, make sure it exists at "
                f"`{expected_file_path}.py`."
            )
        else:
            raise ImportError(
                f"Couldn't import flavor {flavor_model.name}: {err}"
            )
    return cast(Flavor, flavor)
generate_default_docs_url() -> str

Generate the doc urls for all inbuilt and integration flavors.

Note that this method is not going to be useful for custom flavors, which do not have any docs in the main zenml docs.

Returns:

Type Description
str

The complete url to the zenml documentation
Source code in src/zenml/stack/flavor.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def generate_default_docs_url(self) -> str:
    """Generate the doc urls for all inbuilt and integration flavors.

    Note that this method is not going to be useful for custom flavors,
    which do not have any docs in the main zenml docs.

    Returns:
        The complete url to the zenml documentation
    """
    component_type = self.type.plural.replace("_", "-")
    name = self.name.replace("_", "-")

    base = "https://docs.zenml.io"
    return f"{base}/stack-components/{component_type}/{name}"
generate_default_sdk_docs_url() -> str

Generate SDK docs url for a flavor.

Returns:

Type Description
str

The complete url to the zenml SDK docs
Source code in src/zenml/stack/flavor.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def generate_default_sdk_docs_url(self) -> str:
    """Generate SDK docs url for a flavor.

    Returns:
        The complete url to the zenml SDK docs
    """
    from zenml import __version__

    base = f"https://sdkdocs.zenml.io/{__version__}"

    component_type = self.type.plural

    if "zenml.integrations" in self.__module__:
        # Get integration name out of module path which will look something
        #  like this "zenml.integrations.<integration>....
        integration = self.__module__.split(
            "zenml.integrations.", maxsplit=1
        )[1].split(".")[0]

        # Get the config class name to point to the specific class
        config_class_name = self.config_class.__name__

        return (
            f"{base}/integration_code_docs"
            f"/integrations-{integration}"
            f"#zenml.integrations.{integration}.flavors.{config_class_name}"
        )

    else:
        return (
            f"{base}/core_code_docs/core-{component_type}/"
            f"#{self.__module__}"
        )
to_model(integration: Optional[str] = None, is_custom: bool = True) -> FlavorRequest

Converts a flavor to a model.

Parameters:

Name Type Description Default
integration Optional[str]

The integration to use for the model.

 None
is_custom bool

Whether the flavor is a custom flavor.

 True

Returns:

Type Description
FlavorRequest

The model.
Source code in src/zenml/stack/flavor.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def to_model(
    self,
    integration: Optional[str] = None,
    is_custom: bool = True,
) -> FlavorRequest:
    """Converts a flavor to a model.

    Args:
        integration: The integration to use for the model.
        is_custom: Whether the flavor is a custom flavor.

    Returns:
        The model.
    """
    connector_requirements = self.service_connector_requirements
    connector_type = (
        connector_requirements.connector_type
        if connector_requirements
        else None
    )
    resource_type = (
        connector_requirements.resource_type
        if connector_requirements
        else None
    )
    resource_id_attr = (
        connector_requirements.resource_id_attr
        if connector_requirements
        else None
    )

    model = FlavorRequest(
        name=self.name,
        display_name=self.display_name,
        type=self.type,
        source=source_utils.resolve(self.__class__).import_path,
        config_schema=self.config_schema,
        connector_type=connector_type,
        connector_resource_type=resource_type,
        connector_resource_id_attr=resource_id_attr,
        integration=integration,
        logo_url=self.logo_url,
        docs_url=self.docs_url,
        sdk_docs_url=self.sdk_docs_url,
        is_custom=is_custom,
    )
    return model

Stack(id: UUID, name: str, *, environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None, orchestrator: BaseOrchestrator, artifact_store: BaseArtifactStore, container_registry: Optional[BaseContainerRegistry] = None, step_operator: Optional[Union[BaseStepOperator, List[BaseStepOperator]]] = None, feature_store: Optional[BaseFeatureStore] = None, model_deployer: Optional[BaseModelDeployer] = None, experiment_tracker: Optional[Union[BaseExperimentTracker, List[BaseExperimentTracker]]] = None, alerter: Optional[Union[BaseAlerter, List[BaseAlerter]]] = None, annotator: Optional[BaseAnnotator] = None, data_validator: Optional[BaseDataValidator] = None, image_builder: Optional[BaseImageBuilder] = None, model_registry: Optional[BaseModelRegistry] = None, deployer: Optional[BaseDeployer] = None, log_store: Optional[BaseLogStore] = None)

ZenML stack class.

A ZenML stack is a collection of multiple stack components that are required to run ZenML pipelines. Some of these components (orchestrator, and artifact store) are required to run any kind of pipeline, other components like the container registry are only required if other stack components depend on them.

Initializes and validates a stack instance.

Parameters:

Name Type Description Default
id UUID

Unique ID of the stack.

 required
name str

Name of the stack.

 required
environment Optional[Dict[str, str]]

Environment variables to set when running on this stack.

 None
secrets Optional[List[UUID]]

Secrets to set as environment variables when running on this stack.

 None
orchestrator BaseOrchestrator

Orchestrator component of the stack.

 required
artifact_store BaseArtifactStore

Artifact store component of the stack.

 required
container_registry Optional[BaseContainerRegistry]

Container registry component of the stack.

 None
step_operator Optional[Union[BaseStepOperator, List[BaseStepOperator]]]

Step operator component of the stack.

 None
feature_store Optional[BaseFeatureStore]

Feature store component of the stack.

 None
model_deployer Optional[BaseModelDeployer]

Model deployer component of the stack.

 None
experiment_tracker Optional[Union[BaseExperimentTracker, List[BaseExperimentTracker]]]

Experiment tracker component of the stack.

 None
alerter Optional[Union[BaseAlerter, List[BaseAlerter]]]

Alerter component of the stack.

 None
annotator Optional[BaseAnnotator]

Annotator component of the stack.

 None
data_validator Optional[BaseDataValidator]

Data validator component of the stack.

 None
image_builder Optional[BaseImageBuilder]

Image builder component of the stack.

 None
model_registry Optional[BaseModelRegistry]

Model registry component of the stack.

 None
deployer Optional[BaseDeployer]

Deployer component of the stack.

 None
log_store Optional[BaseLogStore]

Log store component of the stack.

 None
Source code in src/zenml/stack/stack.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
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
153
154
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
181
182
183
184
def __init__(
    self,
    id: UUID,
    name: str,
    *,
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
    orchestrator: "BaseOrchestrator",
    artifact_store: "BaseArtifactStore",
    container_registry: Optional["BaseContainerRegistry"] = None,
    step_operator: Optional[
        Union["BaseStepOperator", List["BaseStepOperator"]]
    ] = None,
    feature_store: Optional["BaseFeatureStore"] = None,
    model_deployer: Optional["BaseModelDeployer"] = None,
    experiment_tracker: Optional[
        Union["BaseExperimentTracker", List["BaseExperimentTracker"]]
    ] = None,
    alerter: Optional[Union["BaseAlerter", List["BaseAlerter"]]] = None,
    annotator: Optional["BaseAnnotator"] = None,
    data_validator: Optional["BaseDataValidator"] = None,
    image_builder: Optional["BaseImageBuilder"] = None,
    model_registry: Optional["BaseModelRegistry"] = None,
    deployer: Optional["BaseDeployer"] = None,
    log_store: Optional["BaseLogStore"] = None,
):
    """Initializes and validates a stack instance.

    Args:
        id: Unique ID of the stack.
        name: Name of the stack.
        environment: Environment variables to set when running on this
            stack.
        secrets: Secrets to set as environment variables when running on
            this stack.
        orchestrator: Orchestrator component of the stack.
        artifact_store: Artifact store component of the stack.
        container_registry: Container registry component of the stack.
        step_operator: Step operator component of the stack.
        feature_store: Feature store component of the stack.
        model_deployer: Model deployer component of the stack.
        experiment_tracker: Experiment tracker component of the stack.
        alerter: Alerter component of the stack.
        annotator: Annotator component of the stack.
        data_validator: Data validator component of the stack.
        image_builder: Image builder component of the stack.
        model_registry: Model registry component of the stack.
        deployer: Deployer component of the stack.
        log_store: Log store component of the stack.
    """
    self._id = id
    self._name = name
    self._environment = environment or {}
    self._secrets = secrets or []

    self._components: Dict[StackComponentType, List["StackComponent"]] = (
        defaultdict(list)
    )

    for component_input in [
        orchestrator,
        artifact_store,
        container_registry,
        step_operator,
        feature_store,
        model_deployer,
        experiment_tracker,
        alerter,
        annotator,
        data_validator,
        image_builder,
        model_registry,
        deployer,
        log_store,
    ]:
        components: List["StackComponent"] = list()
        if component_input:
            if isinstance(component_input, list):
                components.extend(component_input)
            else:
                components.append(component_input)

            for component in components:
                self._components[component.type].append(component)
Attributes
alerter: Optional[BaseAlerter] property

The alerter of the stack.

Returns:

Type Description
Optional[BaseAlerter]

The alerter of the stack.
alerters: Dict[str, BaseAlerter] property

All attached alerters keyed by component name.

Returns:

Type Description
Dict[str, BaseAlerter]

The stack alerters keyed by their names.
all_components: List[StackComponent] property

All attached components of the stack.

Returns:

Type Description
List[StackComponent]

A flat list of all components attached to the stack.
annotator: Optional[BaseAnnotator] property

The annotator of the stack.

Returns:

Type Description
Optional[BaseAnnotator]

The annotator of the stack.
apt_packages: List[str] property

List of APT package requirements for the stack.

Returns:

Type Description
List[str]

A list of APT package requirements for the stack.
artifact_store: BaseArtifactStore property

The artifact store of the stack.

Returns:

Type Description
BaseArtifactStore

The artifact store of the stack.
component_flavor_metadata: Dict[str, str] property

Flavor metadata for attached components grouped by type.

Returns:

Type Description
Dict[str, str]

A mapping from stack component type values to comma-separated
Dict[str, str]

flavor names. Repeated flavors are only included once and the
Dict[str, str]

component order is preserved.
components: Dict[StackComponentType, StackComponent] property

All components of the stack.

Returns:

Type Description
Dict[StackComponentType, StackComponent]

A dictionary of all components of the stack.
container_registry: Optional[BaseContainerRegistry] property

The container registry of the stack.

Returns:

Type Description
Optional[BaseContainerRegistry]

The container registry of the stack or None if the stack does not
Optional[BaseContainerRegistry]

have a container registry.
data_validator: Optional[BaseDataValidator] property

The data validator of the stack.

Returns:

Type Description
Optional[BaseDataValidator]

The data validator of the stack.
deployer: Optional[BaseDeployer] property

The deployer of the stack.

Returns:

Type Description
Optional[BaseDeployer]

The deployer of the stack.
environment: Dict[str, str] property

Environment variables to set when running on this stack.

Returns:

Type Description
Dict[str, str]

Environment variables to set when running on this stack.
experiment_tracker: Optional[BaseExperimentTracker] property

The experiment tracker of the stack.

Returns:

Type Description
Optional[BaseExperimentTracker]

The experiment tracker of the stack.
experiment_trackers: Dict[str, BaseExperimentTracker] property

All attached experiment trackers keyed by component name.

Returns:

Type Description
Dict[str, BaseExperimentTracker]

The stack experiment trackers keyed by their names.
feature_store: Optional[BaseFeatureStore] property

The feature store of the stack.

Returns:

Type Description
Optional[BaseFeatureStore]

The feature store of the stack.
id: UUID property

The ID of the stack.

Returns:

Type Description
UUID

The ID of the stack.
image_builder: Optional[BaseImageBuilder] property

The image builder of the stack.

Returns:

Type Description
Optional[BaseImageBuilder]

The image builder of the stack.
log_store: BaseLogStore property

The log store of the stack.

Returns:

Type Description
BaseLogStore

The log store of the stack.
model_deployer: Optional[BaseModelDeployer] property

The model deployer of the stack.

Returns:

Type Description
Optional[BaseModelDeployer]

The model deployer of the stack.
model_registry: Optional[BaseModelRegistry] property

The model registry of the stack.

Returns:

Type Description
Optional[BaseModelRegistry]

The model registry of the stack.
name: str property

The name of the stack.

Returns:

Name Type Description
str str

The name of the stack.
orchestrator: BaseOrchestrator property

The orchestrator of the stack.

Returns:

Type Description
BaseOrchestrator

The orchestrator of the stack.
required_secrets: Set[secret_utils.SecretReference] property

All required secrets for this stack.

Returns:

Type Description
Set[SecretReference]

The required secrets of this stack.
requires_remote_server: bool property

If the stack requires a remote ZenServer to run.

This is the case if any code is getting executed remotely. This is the case for both remote orchestrators as well as remote step operators.

Returns:

Type Description
bool

If the stack requires a remote ZenServer to run.
secrets: List[UUID] property

Secrets to set as environment variables when running on this stack.

Returns:

Type Description
List[UUID]

Secrets to set as environment variables when running on this stack.
setting_classes: Dict[str, Type[BaseSettings]] property

Setting classes of all components of this stack.

Returns:

Type Description
Dict[str, Type[BaseSettings]]

All setting classes and their respective keys.
step_operator: Optional[BaseStepOperator] property

The step operator of the stack.

Returns:

Type Description
Optional[BaseStepOperator]

The step operator of the stack.
step_operators: Dict[str, BaseStepOperator] property

All attached step operators keyed by component name.

Returns:

Type Description
Dict[str, BaseStepOperator]

The stack step operators keyed by their names.
Functions
check_local_paths() -> bool

Checks if the stack has local paths.

Returns:

Type Description
bool

True if the stack has local paths, False otherwise.

Raises:

Type Description
ValueError

If the stack has local paths that do not conform to the convention that all local path must be relative to the local stores directory.
Source code in src/zenml/stack/stack.py 
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
def check_local_paths(self) -> bool:
    """Checks if the stack has local paths.

    Returns:
        True if the stack has local paths, False otherwise.

    Raises:
        ValueError: If the stack has local paths that do not conform to
            the convention that all local path must be relative to the
            local stores directory.
    """
    from zenml.config.global_config import GlobalConfiguration

    local_stores_path = GlobalConfiguration().local_stores_path

    # go through all stack components and identify those that advertise
    # a local path where they persist information that they need to be
    # available when running pipelines.
    has_local_paths = False
    for stack_comp in self.all_components:
        local_path = stack_comp.local_path
        if not local_path:
            continue
        # double-check this convention, just in case it wasn't respected
        # as documented in `StackComponent.local_path`
        if not local_path.startswith(local_stores_path):
            raise ValueError(
                f"Local path {local_path} for component "
                f"{stack_comp.name} is not in the local stores "
                f"directory ({local_stores_path})."
            )
        has_local_paths = True

    return has_local_paths
cleanup_step_run(info: StepRunInfo, step_failed: bool) -> None

Cleans up resources after the step run is finished.

Parameters:

Name Type Description Default
info StepRunInfo

Info about the step that was executed.

 required
step_failed bool

Whether the step failed.

 required
Source code in src/zenml/stack/stack.py 
1553
1554
1555
1556
1557
1558
1559
1560
1561
def cleanup_step_run(self, info: "StepRunInfo", step_failed: bool) -> None:
    """Cleans up resources after the step run is finished.

    Args:
        info: Info about the step that was executed.
        step_failed: Whether the step failed.
    """
    for component in self._get_active_components_for_step(info.config):
        component.cleanup_step_run(info=info, step_failed=step_failed)
deploy_pipeline(snapshot: PipelineSnapshotResponse, deployment_name: str, timeout: Optional[int] = None) -> DeploymentResponse

Deploys a pipeline on this stack.

Parameters:

Name Type Description Default
snapshot PipelineSnapshotResponse

The pipeline snapshot.

 required
deployment_name str

The name to use for the deployment.

 required
timeout Optional[int]

The maximum time in seconds to wait for the pipeline to be deployed.

 None

Returns:

Type Description
DeploymentResponse

The deployment response.

Raises:

Type Description
RuntimeError

If the stack does not have a deployer.
Source code in src/zenml/stack/stack.py 
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
def deploy_pipeline(
    self,
    snapshot: "PipelineSnapshotResponse",
    deployment_name: str,
    timeout: Optional[int] = None,
) -> "DeploymentResponse":
    """Deploys a pipeline on this stack.

    Args:
        snapshot: The pipeline snapshot.
        deployment_name: The name to use for the deployment.
        timeout: The maximum time in seconds to wait for the pipeline to be
            deployed.

    Returns:
        The deployment response.

    Raises:
        RuntimeError: If the stack does not have a deployer.
    """
    if not self.deployer:
        raise RuntimeError(
            "The stack does not have a deployer. Please add a "
            "deployer to the stack in order to deploy a pipeline."
        )

    return self.deployer.provision_deployment(
        snapshot=snapshot,
        stack=self,
        deployment_name_or_id=deployment_name,
        timeout=timeout,
    )
dict() -> Dict[str, str]

Converts the stack into a dictionary.

Returns:

Type Description
Dict[str, str]

A dictionary containing the stack components.
Source code in src/zenml/stack/stack.py 
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
def dict(self) -> Dict[str, str]:
    """Converts the stack into a dictionary.

    Returns:
        A dictionary containing the stack components.
    """
    component_dict = {
        component_type.value: json.dumps(
            [
                component.config.model_dump(mode="json")
                for component in components
            ],
            sort_keys=True,
        )
        for component_type, components in self._components.items()
    }
    component_dict.update({"name": self.name})
    return component_dict
from_components(id: UUID, name: str, components: Dict[StackComponentType, StackComponent], environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None) -> Stack classmethod

Creates a stack instance from a dict of stack components.

Parameters:

Name Type Description Default
id UUID

Unique ID of the stack.

 required
name str

The name of the stack.

 required
components Dict[StackComponentType, StackComponent]

The components of the stack.

 required
environment Optional[Dict[str, str]]

Environment variables to set when running on this stack.

 None
secrets Optional[List[UUID]]

Secrets to set as environment variables when running on this stack.

 None

Returns:

Type Description
Stack

A stack instance consisting of the given components.

Raises:

Type Description
TypeError

If a required component is missing or a component doesn't inherit from the expected base class.
Source code in src/zenml/stack/stack.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
@classmethod
def from_components(
    cls,
    id: UUID,
    name: str,
    components: Dict[StackComponentType, "StackComponent"],
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
) -> "Stack":
    """Creates a stack instance from a dict of stack components.

    Args:
        id: Unique ID of the stack.
        name: The name of the stack.
        components: The components of the stack.
        environment: Environment variables to set when running on this
            stack.
        secrets: Secrets to set as environment variables when running on
            this stack.

    Returns:
        A stack instance consisting of the given components.

    Raises:
        TypeError: If a required component is missing or a component
            doesn't inherit from the expected base class.
    """  # noqa: DOC502
    logger.warning(
        "The class method `Stack.from_components(...)` has been deprecated. "
        "Please use `Stack.from_components_v2(...)` instead."
    )

    from zenml.alerter import BaseAlerter
    from zenml.annotators import BaseAnnotator
    from zenml.artifact_stores import BaseArtifactStore
    from zenml.container_registries import BaseContainerRegistry
    from zenml.data_validators import BaseDataValidator
    from zenml.deployers import BaseDeployer
    from zenml.experiment_trackers import BaseExperimentTracker
    from zenml.feature_stores import BaseFeatureStore
    from zenml.image_builders import BaseImageBuilder
    from zenml.log_stores import BaseLogStore
    from zenml.model_deployers import BaseModelDeployer
    from zenml.model_registries import BaseModelRegistry
    from zenml.orchestrators import BaseOrchestrator
    from zenml.step_operators import BaseStepOperator

    def _raise_type_error(
        component: Optional["StackComponent"], expected_class: Type[Any]
    ) -> NoReturn:
        """Raises a TypeError that the component has an unexpected type.

        Args:
            component: The component that has an unexpected type.
            expected_class: The expected type of the component.

        Raises:
            TypeError: If the component has an unexpected type.
        """
        raise TypeError(
            f"Unable to create stack: Wrong stack component type "
            f"`{component.__class__.__name__}` (expected: subclass "
            f"of `{expected_class.__name__}`)"
        )

    orchestrator = components.get(StackComponentType.ORCHESTRATOR)
    if not isinstance(orchestrator, BaseOrchestrator):
        _raise_type_error(orchestrator, BaseOrchestrator)

    artifact_store = components.get(StackComponentType.ARTIFACT_STORE)
    if not isinstance(artifact_store, BaseArtifactStore):
        _raise_type_error(artifact_store, BaseArtifactStore)

    container_registry = components.get(
        StackComponentType.CONTAINER_REGISTRY
    )
    if container_registry is not None and not isinstance(
        container_registry, BaseContainerRegistry
    ):
        _raise_type_error(container_registry, BaseContainerRegistry)

    step_operator = components.get(StackComponentType.STEP_OPERATOR)
    if step_operator is not None and not isinstance(
        step_operator, BaseStepOperator
    ):
        _raise_type_error(step_operator, BaseStepOperator)

    feature_store = components.get(StackComponentType.FEATURE_STORE)
    if feature_store is not None and not isinstance(
        feature_store, BaseFeatureStore
    ):
        _raise_type_error(feature_store, BaseFeatureStore)

    model_deployer = components.get(StackComponentType.MODEL_DEPLOYER)
    if model_deployer is not None and not isinstance(
        model_deployer, BaseModelDeployer
    ):
        _raise_type_error(model_deployer, BaseModelDeployer)

    experiment_tracker = components.get(
        StackComponentType.EXPERIMENT_TRACKER
    )
    if experiment_tracker is not None and not isinstance(
        experiment_tracker, BaseExperimentTracker
    ):
        _raise_type_error(experiment_tracker, BaseExperimentTracker)

    alerter = components.get(StackComponentType.ALERTER)
    if alerter is not None and not isinstance(alerter, BaseAlerter):
        _raise_type_error(alerter, BaseAlerter)

    annotator = components.get(StackComponentType.ANNOTATOR)
    if annotator is not None and not isinstance(annotator, BaseAnnotator):
        _raise_type_error(annotator, BaseAnnotator)

    data_validator = components.get(StackComponentType.DATA_VALIDATOR)
    if data_validator is not None and not isinstance(
        data_validator, BaseDataValidator
    ):
        _raise_type_error(data_validator, BaseDataValidator)

    image_builder = components.get(StackComponentType.IMAGE_BUILDER)
    if image_builder is not None and not isinstance(
        image_builder, BaseImageBuilder
    ):
        _raise_type_error(image_builder, BaseImageBuilder)

    model_registry = components.get(StackComponentType.MODEL_REGISTRY)
    if model_registry is not None and not isinstance(
        model_registry, BaseModelRegistry
    ):
        _raise_type_error(model_registry, BaseModelRegistry)

    deployer = components.get(StackComponentType.DEPLOYER)
    if deployer is not None and not isinstance(deployer, BaseDeployer):
        _raise_type_error(deployer, BaseDeployer)

    log_store = components.get(StackComponentType.LOG_STORE)
    if log_store is not None and not isinstance(log_store, BaseLogStore):
        _raise_type_error(log_store, BaseLogStore)

    return Stack(
        id=id,
        name=name,
        environment=environment,
        secrets=secrets,
        orchestrator=orchestrator,
        artifact_store=artifact_store,
        container_registry=container_registry,
        step_operator=step_operator,
        feature_store=feature_store,
        model_deployer=model_deployer,
        experiment_tracker=experiment_tracker,
        alerter=alerter,
        annotator=annotator,
        data_validator=data_validator,
        image_builder=image_builder,
        model_registry=model_registry,
        deployer=deployer,
        log_store=log_store,
    )
from_components_v2(id: UUID, name: str, components: Mapping[StackComponentType, List[StackComponent]], environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None) -> Stack classmethod

Creates a stack instance from a dict of stack components.

Parameters:

Name Type Description Default
id UUID

Unique ID of the stack.

 required
name str

The name of the stack.

 required
components Mapping[StackComponentType, List[StackComponent]]

The components of the stack grouped by type. Callers may pass either a single component per type or a list. The first component in a repeatable component list is treated as the default component for that type.

 required
environment Optional[Dict[str, str]]

Environment variables to set when running on this stack.

 None
secrets Optional[List[UUID]]

Secrets to set as environment variables when running on this stack.

 None

Returns:

Type Description
Stack

A stack instance consisting of the given components.

Raises:

Type Description
TypeError

If a required component is missing or a component doesn't inherit from the expected base class.
Source code in src/zenml/stack/stack.py 
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
@classmethod
def from_components_v2(
    cls,
    id: UUID,
    name: str,
    components: Mapping[StackComponentType, List["StackComponent"]],
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
) -> "Stack":
    """Creates a stack instance from a dict of stack components.

    Args:
        id: Unique ID of the stack.
        name: The name of the stack.
        components: The components of the stack grouped by type. Callers may
            pass either a single component per type or a list. The first
            component in a repeatable component list is treated as the
            default component for that type.
        environment: Environment variables to set when running on this
            stack.
        secrets: Secrets to set as environment variables when running on
            this stack.

    Returns:
        A stack instance consisting of the given components.

    Raises:
        TypeError: If a required component is missing or a component
            doesn't inherit from the expected base class.
    """  # noqa: DOC502
    from zenml.alerter import BaseAlerter
    from zenml.annotators import BaseAnnotator
    from zenml.artifact_stores import BaseArtifactStore
    from zenml.container_registries import BaseContainerRegistry
    from zenml.data_validators import BaseDataValidator
    from zenml.deployers import BaseDeployer
    from zenml.experiment_trackers import BaseExperimentTracker
    from zenml.feature_stores import BaseFeatureStore
    from zenml.image_builders import BaseImageBuilder
    from zenml.log_stores import BaseLogStore
    from zenml.model_deployers import BaseModelDeployer
    from zenml.model_registries import BaseModelRegistry
    from zenml.orchestrators import BaseOrchestrator
    from zenml.step_operators import BaseStepOperator

    def _raise_type_error(
        component: Optional["StackComponent"], expected_class: Type[Any]
    ) -> NoReturn:
        """Raises a TypeError that the component has an unexpected type.

        Args:
            component: The component that has an unexpected type.
            expected_class: The expected type of the component.

        Raises:
            TypeError: If the component has an unexpected type.
        """
        raise TypeError(
            f"Unable to create stack: Wrong stack component type "
            f"`{component.__class__.__name__}` (expected: subclass "
            f"of `{expected_class.__name__}`)"
        )

    # Orchestrator
    orchestrator = components.get(StackComponentType.ORCHESTRATOR)
    if not orchestrator:
        raise TypeError("No orchestrator found in the stack.")
    if len(orchestrator) > 1:
        raise TypeError("Multiple orchestrators are not supported.")

    if not isinstance(orchestrator[0], BaseOrchestrator):
        _raise_type_error(orchestrator[0], BaseOrchestrator)

    # Artifact Store
    artifact_store = components.get(StackComponentType.ARTIFACT_STORE)
    if not artifact_store:
        raise TypeError("No artifact store found in the stack.")
    if len(artifact_store) > 1:
        raise TypeError("Multiple artifact stores are not supported.")

    if not isinstance(artifact_store[0], BaseArtifactStore):
        _raise_type_error(artifact_store[0], BaseArtifactStore)

    # Container Registry
    container_registry = components.get(
        StackComponentType.CONTAINER_REGISTRY
    )
    if container_registry:
        if len(container_registry) > 1:
            raise TypeError(
                "Multiple container registries are not supported."
            )
        if not isinstance(container_registry[0], BaseContainerRegistry):
            _raise_type_error(container_registry[0], BaseContainerRegistry)

    # Step Operator
    step_operator = components.get(StackComponentType.STEP_OPERATOR)
    if step_operator is not None and not isinstance(step_operator, list):
        step_operator = [step_operator]
    for so in step_operator or []:
        if not isinstance(so, BaseStepOperator):
            _raise_type_error(so, BaseStepOperator)

    # Feature Store
    feature_store = components.get(StackComponentType.FEATURE_STORE)
    if feature_store:
        if len(feature_store) > 1:
            raise TypeError("Multiple feature stores are not supported.")

        if not isinstance(feature_store[0], BaseFeatureStore):
            _raise_type_error(feature_store[0], BaseFeatureStore)

    # Model Deployer
    model_deployer = components.get(StackComponentType.MODEL_DEPLOYER)
    if model_deployer:
        if len(model_deployer) > 1:
            raise TypeError("Multiple model deployers are not supported.")
        if not isinstance(model_deployer[0], BaseModelDeployer):
            _raise_type_error(model_deployer[0], BaseModelDeployer)

    # Experiment Tracker
    experiment_tracker = components.get(
        StackComponentType.EXPERIMENT_TRACKER
    )
    if experiment_tracker is not None and not isinstance(
        experiment_tracker, list
    ):
        experiment_tracker = [experiment_tracker]

    for et in experiment_tracker or []:
        if not isinstance(et, BaseExperimentTracker):
            _raise_type_error(et, BaseExperimentTracker)

    # Alerter
    alerter = components.get(StackComponentType.ALERTER)
    if alerter is not None and not isinstance(alerter, list):
        alerter = [alerter]
    for a in alerter or []:
        if not isinstance(a, BaseAlerter):
            _raise_type_error(a, BaseAlerter)

    # Annotator
    annotator = components.get(StackComponentType.ANNOTATOR)
    if annotator:
        if len(annotator) > 1:
            raise TypeError("Multiple annotators are not supported.")
        if not isinstance(annotator[0], BaseAnnotator):
            _raise_type_error(annotator[0], BaseAnnotator)

    # Data Validator
    data_validator = components.get(StackComponentType.DATA_VALIDATOR)
    if data_validator:
        if len(data_validator) > 1:
            raise TypeError("Multiple data validators are not supported.")
        if not isinstance(data_validator[0], BaseDataValidator):
            _raise_type_error(data_validator[0], BaseDataValidator)

    # Image Builder
    image_builder = components.get(StackComponentType.IMAGE_BUILDER)
    if image_builder:
        if len(image_builder) > 1:
            raise TypeError("Multiple image builders are not supported.")

        if not isinstance(image_builder[0], BaseImageBuilder):
            _raise_type_error(image_builder[0], BaseImageBuilder)

    # Model Registry
    model_registry = components.get(StackComponentType.MODEL_REGISTRY)
    if model_registry:
        if len(model_registry) > 1:
            raise TypeError("Multiple model registries are not supported.")
        if not isinstance(model_registry[0], BaseModelRegistry):
            _raise_type_error(model_registry[0], BaseModelRegistry)

    # Deployer
    deployer = components.get(StackComponentType.DEPLOYER)
    if deployer:
        if len(deployer) > 1:
            raise TypeError("Multiple deployers are not supported.")
        if not isinstance(deployer[0], BaseDeployer):
            _raise_type_error(deployer[0], BaseDeployer)

    # Log Store
    log_store = components.get(StackComponentType.LOG_STORE)
    if log_store:
        if len(log_store) > 1:
            raise TypeError("Multiple log stores are not supported.")
        if not isinstance(log_store[0], BaseLogStore):
            _raise_type_error(log_store[0], BaseLogStore)

    return Stack(
        id=id,
        name=name,
        environment=environment,
        secrets=secrets,
        orchestrator=orchestrator[0],
        artifact_store=artifact_store[0],
        container_registry=cast(
            Optional["BaseContainerRegistry"],
            container_registry[0] if container_registry else None,
        ),
        step_operator=cast(
            Optional[Union["BaseStepOperator", List["BaseStepOperator"]]],
            step_operator,
        ),
        feature_store=cast(
            Optional["BaseFeatureStore"],
            feature_store[0] if feature_store else None,
        ),
        model_deployer=cast(
            Optional["BaseModelDeployer"],
            model_deployer[0] if model_deployer else None,
        ),
        experiment_tracker=cast(
            Optional[
                Union[
                    "BaseExperimentTracker",
                    List["BaseExperimentTracker"],
                ]
            ],
            experiment_tracker,
        ),
        alerter=cast(
            Optional[Union["BaseAlerter", List["BaseAlerter"]]],
            alerter,
        ),
        annotator=cast(
            Optional["BaseAnnotator"], annotator[0] if annotator else None
        ),
        data_validator=cast(
            Optional["BaseDataValidator"],
            data_validator[0] if data_validator else None,
        ),
        image_builder=cast(
            Optional["BaseImageBuilder"],
            image_builder[0] if image_builder else None,
        ),
        model_registry=cast(
            Optional["BaseModelRegistry"],
            model_registry[0] if model_registry else None,
        ),
        deployer=cast(
            Optional["BaseDeployer"], deployer[0] if deployer else None
        ),
        log_store=cast(
            Optional["BaseLogStore"], log_store[0] if log_store else None
        ),
    )
from_model(stack_model: StackResponse) -> Stack classmethod

Creates a Stack instance from a StackModel.

Parameters:

Name Type Description Default
stack_model StackResponse

The StackModel to create the Stack from.

 required

Returns:

Type Description
Stack

The created Stack instance.

Raises:

Type Description
ImportError

If a stack component's dependencies cannot be imported.
Source code in src/zenml/stack/stack.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
@classmethod
def from_model(cls, stack_model: "StackResponse") -> "Stack":
    """Creates a Stack instance from a StackModel.

    Args:
        stack_model: The StackModel to create the Stack from.

    Returns:
        The created Stack instance.

    Raises:
        ImportError: If a stack component's dependencies cannot be imported.
    """
    global _STACK_CACHE
    key = (stack_model.id, stack_model.updated)
    if key in _STACK_CACHE:
        return _STACK_CACHE[key]

    from zenml.stack import StackComponent

    # Run a hydrated list call once to avoid one request per component
    hydrated_component_models = pagination_utils.depaginate(
        Client().list_stack_components,
        stack_id=stack_model.id,
        hydrate=True,
    )

    try:
        stack_components: Dict[
            StackComponentType, List["StackComponent"]
        ] = defaultdict(list)

        for component in hydrated_component_models:
            stack_components[component.type].append(
                StackComponent.from_model(component)
            )
    except ImportError as e:
        stack_hint = (
            "\n\nTo install all requirements for this stack at "
            "once, run:\n\n"
            f"  zenml stack export-requirements "
            f"'{stack_model.name}' -o stack-requirements.txt\n"
            "  pip install -r stack-requirements.txt"
        )
        e.args = (e.args[0] + stack_hint,) + e.args[1:]
        raise

    stack = Stack.from_components_v2(
        id=stack_model.id,
        name=stack_model.name,
        environment=stack_model.environment,
        secrets=stack_model.secrets,
        components=stack_components,
    )
    _STACK_CACHE[key] = stack

    client = Client()
    if stack_model.id == client.active_stack_model.id:
        if stack_model.updated > client.active_stack_model.updated:
            if client._config:
                client._config.set_active_stack(stack_model)
            else:
                GlobalConfiguration().set_active_stack(stack_model)

    return stack
get_components_by_type(component_type: StackComponentType) -> Sequence[StackComponent]

Returns all components of a given type.

Parameters:

Name Type Description Default
component_type StackComponentType

The component type to fetch.

 required

Returns:

Type Description
Sequence[StackComponent]

A list of attached components of the given type.
Source code in src/zenml/stack/stack.py 
694
695
696
697
698
699
700
701
702
703
704
705
def get_components_by_type(
    self, component_type: StackComponentType
) -> Sequence["StackComponent"]:
    """Returns all components of a given type.

    Args:
        component_type: The component type to fetch.

    Returns:
        A list of attached components of the given type.
    """
    return self._components.get(component_type, [])
get_docker_builds(snapshot: PipelineSnapshotBase) -> List[BuildConfiguration]

Gets the Docker builds required for the stack.

Parameters:

Name Type Description Default
snapshot PipelineSnapshotBase

The pipeline snapshot for which to get the builds.

 required

Returns:

Type Description
List[BuildConfiguration]

The required Docker builds.
Source code in src/zenml/stack/stack.py 
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
def get_docker_builds(
    self, snapshot: "PipelineSnapshotBase"
) -> List["BuildConfiguration"]:
    """Gets the Docker builds required for the stack.

    Args:
        snapshot: The pipeline snapshot for which to get the builds.

    Returns:
        The required Docker builds.
    """
    return list(
        itertools.chain.from_iterable(
            component.get_docker_builds(snapshot=snapshot)
            for component in self.all_components
        )
    )
get_pipeline_run_metadata(run_id: UUID) -> Dict[UUID, Dict[str, MetadataType]]

Get general component-specific metadata for a pipeline run.

Parameters:

Name Type Description Default
run_id UUID

ID of the pipeline run.

 required

Returns:

Type Description
Dict[UUID, Dict[str, MetadataType]]

A dictionary mapping component IDs to the metadata they created.
Source code in src/zenml/stack/stack.py 
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
def get_pipeline_run_metadata(
    self, run_id: UUID
) -> Dict[UUID, Dict[str, MetadataType]]:
    """Get general component-specific metadata for a pipeline run.

    Args:
        run_id: ID of the pipeline run.

    Returns:
        A dictionary mapping component IDs to the metadata they created.
    """
    pipeline_run_metadata: Dict[UUID, Dict[str, MetadataType]] = {}
    for component in self.all_components:
        try:
            component_metadata = component.get_pipeline_run_metadata(
                run_id=run_id
            )
            if component_metadata:
                pipeline_run_metadata[component.id] = component_metadata
        except Exception as e:
            logger.warning(
                f"Extracting pipeline run metadata failed for component "
                f"'{component.name}' of type '{component.type}': {e}"
            )
    return pipeline_run_metadata
get_step_run_metadata(info: StepRunInfo) -> Dict[UUID, Dict[str, MetadataType]]

Get component-specific metadata for a step run.

Parameters:

Name Type Description Default
info StepRunInfo

Info about the step that was executed.

 required

Returns:

Type Description
Dict[UUID, Dict[str, MetadataType]]

A dictionary mapping component IDs to the metadata they created.
Source code in src/zenml/stack/stack.py 
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
def get_step_run_metadata(
    self, info: "StepRunInfo"
) -> Dict[UUID, Dict[str, MetadataType]]:
    """Get component-specific metadata for a step run.

    Args:
        info: Info about the step that was executed.

    Returns:
        A dictionary mapping component IDs to the metadata they created.
    """
    step_run_metadata: Dict[UUID, Dict[str, MetadataType]] = {}
    for component in self._get_active_components_for_step(info.config):
        try:
            component_metadata = component.get_step_run_metadata(info=info)
            if component_metadata:
                step_run_metadata[component.id] = component_metadata
        except Exception as e:
            logger.warning(
                f"Extracting step run metadata failed for component "
                f"'{component.name}' of type '{component.type}': {e}"
            )
    return step_run_metadata
prepare_pipeline_submission(snapshot: PipelineSnapshotResponse) -> None

Prepares the stack for a pipeline submission.

This method is called before a pipeline is submitted.

Parameters:

Name Type Description Default
snapshot PipelineSnapshotResponse

The pipeline snapshot

 required

Raises:

Type Description
RuntimeError

If trying to submit a pipeline that requires a remote ZenML server with a local one.
Source code in src/zenml/stack/stack.py 
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
def prepare_pipeline_submission(
    self, snapshot: "PipelineSnapshotResponse"
) -> None:
    """Prepares the stack for a pipeline submission.

    This method is called before a pipeline is submitted.

    Args:
        snapshot: The pipeline snapshot

    Raises:
        RuntimeError: If trying to submit a pipeline that requires a remote
            ZenML server with a local one.
    """
    self.validate(fail_if_secrets_missing=True)

    if self.requires_remote_server and Client().zen_store.is_local_store():
        raise RuntimeError(
            "Stacks with remote components such as remote orchestrators "
            "and step operators require a remote "
            "ZenML server. To run a pipeline with this stack you need to "
            "connect to a remote ZenML server first. Check out "
            "https://docs.zenml.io/getting-started/deploying-zenml "
            "for more information on how to deploy ZenML."
        )
prepare_step_run(info: StepRunInfo) -> None

Prepares running a step.

Parameters:

Name Type Description Default
info StepRunInfo

Info about the step that will be executed.

 required
Source code in src/zenml/stack/stack.py 
1494
1495
1496
1497
1498
1499
1500
1501
def prepare_step_run(self, info: "StepRunInfo") -> None:
    """Prepares running a step.

    Args:
        info: Info about the step that will be executed.
    """
    for component in self._get_active_components_for_step(info.config):
        component.prepare_step_run(info=info)
requirements(exclude_components: Optional[AbstractSet[StackComponentType]] = None) -> Set[str]

Set of PyPI requirements for the stack.

This method combines the requirements of all stack components (except the ones specified in exclude_components).

Parameters:

Name Type Description Default
exclude_components Optional[AbstractSet[StackComponentType]]

Set of component types for which the requirements should not be included in the output.

 None

Returns:

Type Description
Set[str]

Set of PyPI requirements.
Source code in src/zenml/stack/stack.py 
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
def requirements(
    self,
    exclude_components: Optional[AbstractSet[StackComponentType]] = None,
) -> Set[str]:
    """Set of PyPI requirements for the stack.

    This method combines the requirements of all stack components (except
    the ones specified in `exclude_components`).

    Args:
        exclude_components: Set of component types for which the
            requirements should not be included in the output.

    Returns:
        Set of PyPI requirements.
    """
    exclude_components = exclude_components or set()
    requirements = [
        component.requirements
        for component in self.all_components
        if component.type not in exclude_components
    ]
    return set.union(*requirements) if requirements else set()
submit_pipeline(snapshot: PipelineSnapshotResponse, placeholder_run: Optional[PipelineRunResponse] = None) -> None

Submits a pipeline on this stack.

Parameters:

Name Type Description Default
snapshot PipelineSnapshotResponse

The pipeline snapshot.

 required
placeholder_run Optional[PipelineRunResponse]

An optional placeholder run for the snapshot.

 None
Source code in src/zenml/stack/stack.py 
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
def submit_pipeline(
    self,
    snapshot: "PipelineSnapshotResponse",
    placeholder_run: Optional["PipelineRunResponse"] = None,
) -> None:
    """Submits a pipeline on this stack.

    Args:
        snapshot: The pipeline snapshot.
        placeholder_run: An optional placeholder run for the snapshot.
    """
    self.orchestrator.run(
        snapshot=snapshot, stack=self, placeholder_run=placeholder_run
    )
validate(fail_if_secrets_missing: bool = False) -> None

Checks whether the stack configuration is valid.

To check if a stack configuration is valid, the following criteria must be met: - the stack must have an image builder if other components require it - the StackValidator of each stack component has to validate the stack to make sure all the components are compatible with each other - the required secrets of all components need to exist

Parameters:

Name Type Description Default
fail_if_secrets_missing bool

If this is True, an error will be raised if a secret for a component is missing. Otherwise, only a warning will be logged.

 False
Source code in src/zenml/stack/stack.py 
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
def validate(
    self,
    fail_if_secrets_missing: bool = False,
) -> None:
    """Checks whether the stack configuration is valid.

    To check if a stack configuration is valid, the following criteria must
    be met:
    - the stack must have an image builder if other components require it
    - the `StackValidator` of each stack component has to validate the
        stack to make sure all the components are compatible with each other
    - the required secrets of all components need to exist

    Args:
        fail_if_secrets_missing: If this is `True`, an error will be raised
            if a secret for a component is missing. Otherwise, only a
            warning will be logged.
    """
    if handle_bool_env_var(ENV_ZENML_SKIP_STACK_VALIDATION, default=False):
        logger.debug("Skipping stack validation.")
        return

    self.validate_image_builder()
    self.validate_log_store()
    for component in self.all_components:
        if component.validator:
            component.validator.validate(stack=self)

    self._validate_secrets(raise_exception=fail_if_secrets_missing)
validate_image_builder() -> None

Validates that the stack has an image builder if required.

If the stack requires an image builder, but none is specified, a local image builder will be created and assigned to the stack to ensure backwards compatibility.

Source code in src/zenml/stack/stack.py 
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
def validate_image_builder(self) -> None:
    """Validates that the stack has an image builder if required.

    If the stack requires an image builder, but none is specified, a
    local image builder will be created and assigned to the stack to
    ensure backwards compatibility.
    """
    requires_image_builder = (
        self.orchestrator.flavor != "local"
        or self.step_operator
        or self.deployer
        or (self.model_deployer and self.model_deployer.flavor != "mlflow")
    )
    skip_default_image_builder = handle_bool_env_var(
        ENV_ZENML_SKIP_IMAGE_BUILDER_DEFAULT, default=False
    )
    if (
        requires_image_builder
        and not skip_default_image_builder
        and not self.image_builder
    ):
        from uuid import uuid4

        from zenml.image_builders import (
            LocalImageBuilder,
            LocalImageBuilderConfig,
            LocalImageBuilderFlavor,
        )

        flavor = LocalImageBuilderFlavor()

        now = utc_now()
        image_builder = LocalImageBuilder(
            id=uuid4(),
            name="temporary_default",
            flavor=flavor.name,
            type=flavor.type,
            config=LocalImageBuilderConfig(),
            environment={},
            user=Client().active_user.id,
            created=now,
            updated=now,
            secrets=[],
        )

        self._components[StackComponentType.IMAGE_BUILDER] = [
            image_builder
        ]
validate_log_store() -> None

Validates that the stack has a log store.

Source code in src/zenml/stack/stack.py 
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
def validate_log_store(self) -> None:
    """Validates that the stack has a log store."""
    if self._get_default_component(StackComponentType.LOG_STORE):
        return

    from zenml.log_stores import ArtifactLogStore

    self._components[StackComponentType.LOG_STORE] = [
        ArtifactLogStore.from_artifact_store(self.artifact_store),
    ]

StackComponent(name: str, id: UUID, config: StackComponentConfig, flavor: str, type: StackComponentType, user: Optional[UUID], created: datetime, updated: datetime, environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None, labels: Optional[Dict[str, Any]] = None, connector_requirements: Optional[ServiceConnectorRequirements] = None, connector: Optional[UUID] = None, connector_resource_id: Optional[str] = None, *args: Any, **kwargs: Any)

Abstract StackComponent class for all components of a ZenML stack.

Initializes a StackComponent.

Parameters:

Name Type Description Default
name str

The name of the component.

 required
id UUID

The unique ID of the component.

 required
config StackComponentConfig

The config of the component.

 required
flavor str

The flavor of the component.

 required
type StackComponentType

The type of the component.

 required
user Optional[UUID]

The ID of the user who created the component.

 required
created datetime

The creation time of the component.

 required
updated datetime

The last update time of the component.

 required
environment Optional[Dict[str, str]]

Environment variables to set when running on this component.

 None
secrets Optional[List[UUID]]

Secrets to set as environment variables when running on this component.

 None
labels Optional[Dict[str, Any]]

The labels of the component.

 None
connector_requirements Optional[ServiceConnectorRequirements]

The requirements for the connector.

 None
connector Optional[UUID]

The ID of a connector linked to the component.

 None
connector_resource_id Optional[str]

The custom resource ID to access through the connector.

 None
*args Any

Additional positional arguments.

 ()
**kwargs Any

Additional keyword arguments.

 {}

Raises:

Type Description
ValueError

If a secret reference is passed as name.
Source code in src/zenml/stack/stack_component.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
def __init__(
    self,
    name: str,
    id: UUID,
    config: StackComponentConfig,
    flavor: str,
    type: StackComponentType,
    user: Optional[UUID],
    created: datetime,
    updated: datetime,
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
    labels: Optional[Dict[str, Any]] = None,
    connector_requirements: Optional[ServiceConnectorRequirements] = None,
    connector: Optional[UUID] = None,
    connector_resource_id: Optional[str] = None,
    *args: Any,
    **kwargs: Any,
):
    """Initializes a StackComponent.

    Args:
        name: The name of the component.
        id: The unique ID of the component.
        config: The config of the component.
        flavor: The flavor of the component.
        type: The type of the component.
        user: The ID of the user who created the component.
        created: The creation time of the component.
        updated: The last update time of the component.
        environment: Environment variables to set when running on this
            component.
        secrets: Secrets to set as environment variables when running on
            this component.
        labels: The labels of the component.
        connector_requirements: The requirements for the connector.
        connector: The ID of a connector linked to the component.
        connector_resource_id: The custom resource ID to access through
            the connector.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Raises:
        ValueError: If a secret reference is passed as name.
    """
    if secret_utils.is_secret_reference(name):
        raise ValueError(
            "Passing the `name` attribute of a stack component as a "
            "secret reference is not allowed."
        )

    self.id = id
    self.name = name
    self._config = config
    self.flavor = flavor
    self.type = type
    self.user = user
    self.created = created
    self.updated = updated
    self.labels = labels
    self.environment = environment or {}
    self.secrets = secrets or []
    self.connector_requirements = connector_requirements
    self.connector = connector
    self.connector_resource_id = connector_resource_id
    self._connector_instance: Optional[ServiceConnector] = None
Attributes
apt_packages: List[str] property

List of APT package requirements for the component.

Returns:

Type Description
List[str]

A list of APT package requirements for the component.
config: StackComponentConfig property

Returns the configuration of the stack component.

This should be overwritten by any subclasses that define custom configs to return the correct config class.

Returns:

Type Description
StackComponentConfig

The configuration of the stack component.
local_path: Optional[str] property

Path to a local directory to store persistent information.

This property should only be implemented by components that need to store persistent information in a directory on the local machine and also need that information to be available during pipeline runs.

IMPORTANT: the path returned by this property must always be a path that is relative to the ZenML local store's directory. The local orchestrators rely on this convention to correctly mount the local folders in the containers. This is an example of a valid path:

from zenml.config.global_config import GlobalConfiguration

...

@property
def local_path(self) -> Optional[str]:

    return os.path.join(
        GlobalConfiguration().local_stores_path,
        str(self.uuid),
    )

Returns:

Type Description
Optional[str]

A path to a local directory used by the component to store
Optional[str]

persistent information.
log_file: Optional[str] property

Optional path to a log file for the stack component.

Returns:

Type Description
Optional[str]

Optional path to a log file for the stack component.
post_registration_message: Optional[str] property

Optional message printed after the stack component is registered.

Returns:

Type Description
Optional[str]

An optional message.
requirements: Set[str] property

Set of PyPI requirements for the component.

Returns:

Type Description
Set[str]

A set of PyPI requirements for the component.
settings_class: Optional[Type[BaseSettings]] property

Class specifying available settings for this component.

Returns:

Type Description
Optional[Type[BaseSettings]]

Optional settings class.
validator: Optional[StackValidator] property

The optional validator of the stack component.

This validator will be called each time a stack with the stack component is initialized. Subclasses should override this property and return a StackValidator that makes sure they're not included in any stack that they're not compatible with.

Returns:

Type Description
Optional[StackValidator]

An optional StackValidator instance.
Functions
cleanup() -> None

Cleans up the component after it has been used.

Source code in src/zenml/stack/stack_component.py 
794
795
796
def cleanup(self) -> None:
    """Cleans up the component after it has been used."""
    pass
cleanup_step_run(info: StepRunInfo, step_failed: bool) -> None

Cleans up resources after the step run is finished.

Parameters:

Name Type Description Default
info StepRunInfo

Info about the step that was executed.

 required
step_failed bool

Whether the step failed.

 required
Source code in src/zenml/stack/stack_component.py 
763
764
765
766
767
768
769
def cleanup_step_run(self, info: "StepRunInfo", step_failed: bool) -> None:
    """Cleans up resources after the step run is finished.

    Args:
        info: Info about the step that was executed.
        step_failed: Whether the step failed.
    """
connector_has_expired() -> bool

Checks whether the connector linked to this stack component has expired.

Returns:

Type Description
bool

Whether the connector linked to this stack component has expired, or isn't linked to a connector.
Source code in src/zenml/stack/stack_component.py 
562
563
564
565
566
567
568
569
570
571
572
573
574
575
def connector_has_expired(self) -> bool:
    """Checks whether the connector linked to this stack component has expired.

    Returns:
        Whether the connector linked to this stack component has expired, or isn't linked to a connector.
    """
    if self.connector is None:
        # The stack component isn't linked to a connector
        return False

    if self._connector_instance is None:
        return True

    return self._connector_instance.has_expired()
from_model(component_model: ComponentResponse) -> StackComponent classmethod

Creates a StackComponent from a ComponentModel.

Parameters:

Name Type Description Default
component_model ComponentResponse

The ComponentModel to create the StackComponent

 required

Returns:

Type Description
StackComponent

The created StackComponent.

Raises:

Type Description
ImportError

If the flavor can't be imported.
Source code in src/zenml/stack/stack_component.py 
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
@classmethod
def from_model(
    cls, component_model: "ComponentResponse"
) -> "StackComponent":
    """Creates a StackComponent from a ComponentModel.

    Args:
        component_model: The ComponentModel to create the StackComponent

    Returns:
        The created StackComponent.

    Raises:
        ImportError: If the flavor can't be imported.
    """
    from zenml.stack import Flavor

    flavor_model = component_model.flavor
    flavor = Flavor.from_model(flavor_model)

    configuration = flavor.config_class(**component_model.configuration)

    user_id = component_model.user_id

    try:
        return flavor.implementation_class(
            user=user_id,
            name=component_model.name,
            id=component_model.id,
            config=configuration,
            environment=component_model.environment,
            secrets=component_model.secrets,
            labels=component_model.labels,
            flavor=component_model.flavor_name,
            type=component_model.type,
            created=component_model.created,
            updated=component_model.updated,
            connector_requirements=flavor.service_connector_requirements,
            connector=component_model.connector.id
            if component_model.connector
            else None,
            connector_resource_id=component_model.connector_resource_id,
        )
    except ImportError as e:
        from zenml.integrations.registry import integration_registry

        integration_requirements = " ".join(
            f"'{req}'"
            for req in integration_registry.select_integration_requirements(
                flavor_model.integration
            )
        )

        if integration_registry.is_installed(flavor_model.integration):
            raise ImportError(
                f"{e}\n\n"
                f"Something went wrong while trying to import from the "
                f"`{flavor_model.integration}` integration. Please make "
                "sure that all its requirements are installed properly by "
                "reinstalling the integration either through our CLI: "
                f"`zenml integration install {flavor_model.integration} "
                "-y` or by manually installing its requirements: "
                f"`pip install {integration_requirements}`. If the error "
                "persists, please contact the ZenML team."
            ) from e
        else:
            raise ImportError(
                f"{e}\n\n"
                f"The `{flavor_model.integration}` integration that you "
                "are trying to use is not installed in your current "
                "environment. Please make sure that it is installed by "
                "either using our CLI: `zenml integration install "
                f"{flavor_model.integration}` or by manually installing "
                f"its requirements: `pip install "
                f"{integration_requirements}`"
            ) from e
get_connector() -> Optional[ServiceConnector]

Returns the connector linked to this stack component.

Returns:

Type Description
Optional[ServiceConnector]

The connector linked to this stack component.

Raises:

Type Description
RuntimeError

If the stack component does not specify connector requirements or if the connector linked to the component is not compatible or not found.
Source code in src/zenml/stack/stack_component.py 
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
def get_connector(self) -> Optional["ServiceConnector"]:
    """Returns the connector linked to this stack component.

    Returns:
        The connector linked to this stack component.

    Raises:
        RuntimeError: If the stack component does not specify connector
            requirements or if the connector linked to the component is not
            compatible or not found.
    """
    from zenml.client import Client

    if self.connector is None:
        return None

    if self._connector_instance is not None:
        # If the connector instance is still valid, return it. Otherwise,
        # we'll try to get a new one.
        if not self._connector_instance.has_expired():
            return self._connector_instance

    if self.connector_requirements is None:
        raise RuntimeError(
            f"Unable to get connector for component {self} because this "
            "component does not declare any connector requirements in its. "
            "flavor specification. Override the "
            "`service_connector_requirements` method in its flavor class "
            "to return a connector requirements specification and try "
            "again."
        )

    if self.connector_requirements.resource_id_attr is not None:
        # Check if an attribute is set in the component configuration
        resource_id = getattr(
            self.config, self.connector_requirements.resource_id_attr
        )
    else:
        # Otherwise, use the resource ID configured in the component
        resource_id = self.connector_resource_id

    client = Client()
    try:
        self._connector_instance = client.get_service_connector_client(
            name_id_or_prefix=self.connector,
            resource_type=self.connector_requirements.resource_type,
            resource_id=resource_id,
        )
    except KeyError:
        raise RuntimeError(
            f"The connector with ID {self.connector} linked "
            f"to the '{self.name}' {self.type} stack component could not "
            f"be found or is not accessible. Please verify that the "
            f"connector exists and that you have access to it."
        )
    except ValueError as e:
        raise RuntimeError(
            f"The connector with ID {self.connector} linked "
            f"to the '{self.name}' {self.type} stack component could not "
            f"be correctly configured: {e}."
        )
    except AuthorizationException as e:
        raise RuntimeError(
            f"The connector with ID {self.connector} linked "
            f"to the '{self.name}' {self.type} stack component could not "
            f"be accessed due to an authorization error: {e}. Please "
            f"verify that you have access to the connector and try again."
        )

    return self._connector_instance
get_docker_builds(snapshot: PipelineSnapshotBase) -> List[BuildConfiguration]

Gets the Docker builds required for the component.

Parameters:

Name Type Description Default
snapshot PipelineSnapshotBase

The pipeline snapshot for which to get the builds.

 required

Returns:

Type Description
List[BuildConfiguration]

The required Docker builds.
Source code in src/zenml/stack/stack_component.py 
717
718
719
720
721
722
723
724
725
726
727
728
def get_docker_builds(
    self, snapshot: "PipelineSnapshotBase"
) -> List["BuildConfiguration"]:
    """Gets the Docker builds required for the component.

    Args:
        snapshot: The pipeline snapshot for which to get the builds.

    Returns:
        The required Docker builds.
    """
    return []
get_pipeline_run_metadata(run_id: UUID) -> Dict[str, MetadataType]

Get general component-specific metadata for a pipeline run.

Parameters:

Name Type Description Default
run_id UUID

The ID of the pipeline run.

 required

Returns:

Type Description
Dict[str, MetadataType]

A dictionary of metadata.
Source code in src/zenml/stack/stack_component.py 
730
731
732
733
734
735
736
737
738
739
740
741
def get_pipeline_run_metadata(
    self, run_id: UUID
) -> Dict[str, "MetadataType"]:
    """Get general component-specific metadata for a pipeline run.

    Args:
        run_id: The ID of the pipeline run.

    Returns:
        A dictionary of metadata.
    """
    return {}
get_settings(container: Union[Step, StepRunResponse, StepRunInfo, PipelineSnapshotBase, PipelineSnapshotResponse, PipelineRunResponse]) -> BaseSettings

Gets settings for this stack component.

This will return None if the stack component doesn't specify a settings class or the container doesn't contain runtime options for this component.

Parameters:

Name Type Description Default
container Union[Step, StepRunResponse, StepRunInfo, PipelineSnapshotBase, PipelineSnapshotResponse, PipelineRunResponse]

The Step, StepRunInfo or PipelineSnapshot from which to get the settings.

 required

Returns:

Type Description
BaseSettings

Settings for this stack component.

Raises:

Type Description
RuntimeError

If the stack component does not specify a settings class.
Source code in src/zenml/stack/stack_component.py 
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
def get_settings(
    self,
    container: Union[
        "Step",
        "StepRunResponse",
        "StepRunInfo",
        "PipelineSnapshotBase",
        "PipelineSnapshotResponse",
        "PipelineRunResponse",
    ],
) -> "BaseSettings":
    """Gets settings for this stack component.

    This will return `None` if the stack component doesn't specify a
    settings class or the container doesn't contain runtime
    options for this component.

    Args:
        container: The `Step`, `StepRunInfo` or `PipelineSnapshot` from
            which to get the settings.

    Returns:
        Settings for this stack component.

    Raises:
        RuntimeError: If the stack component does not specify a settings
            class.
    """
    if not self.settings_class:
        raise RuntimeError(
            f"Unable to get settings for component {self} because this "
            "component does not have an associated settings class. "
            "Return a settings class from the `@settings_class` property "
            "and try again."
        )

    canonical_key = settings_utils.get_stack_component_name_setting_key(
        self
    )
    legacy_key = settings_utils.get_stack_component_setting_key(self)

    all_settings = (
        container.config.settings
        if isinstance(
            container,
            (Step, StepRunResponse, StepRunInfo, PipelineRunResponse),
        )
        else container.pipeline_configuration.settings
    )

    # Use the current config as a base
    settings_dict = self.config.model_dump(exclude_unset=True)

    if canonical_key in all_settings:
        settings_dict.update(dict(all_settings[canonical_key]))
    elif legacy_key in all_settings:
        settings_dict.update(dict(all_settings[legacy_key]))

    return self.settings_class.model_validate(settings_dict)
get_step_run_metadata(info: StepRunInfo) -> Dict[str, MetadataType]

Get component- and step-specific metadata after a step ran.

Parameters:

Name Type Description Default
info StepRunInfo

Info about the step that was executed.

 required

Returns:

Type Description
Dict[str, MetadataType]

A dictionary of metadata.
Source code in src/zenml/stack/stack_component.py 
750
751
752
753
754
755
756
757
758
759
760
761
def get_step_run_metadata(
    self, info: "StepRunInfo"
) -> Dict[str, "MetadataType"]:
    """Get component- and step-specific metadata after a step ran.

    Args:
        info: Info about the step that was executed.

    Returns:
        A dictionary of metadata.
    """
    return {}
prepare_step_run(info: StepRunInfo) -> None

Prepares running a step.

Parameters:

Name Type Description Default
info StepRunInfo

Info about the step that will be executed.

 required
Source code in src/zenml/stack/stack_component.py 
743
744
745
746
747
748
def prepare_step_run(self, info: "StepRunInfo") -> None:
    """Prepares running a step.

    Args:
        info: Info about the step that will be executed.
    """

StackComponentConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: BaseModel, ABC

Base class for all ZenML stack component configs.

Ensures that secret references don't clash with pydantic validation.

StackComponents allow the specification of all their string attributes using secret references of the form {{secret_name.key}}. This however is only possible when the stack component does not perform any explicit validation of this attribute using pydantic validators. If this were the case, the validation would run on the secret reference and would fail or in the worst case, modify the secret reference and lead to unexpected behavior. This method ensures that no attributes that require custom pydantic validation are set as secret references.

Parameters:

Name Type Description Default
warn_about_plain_text_secrets bool

If true, then warns about using plain-text secrets.

 False
**kwargs Any

Arguments to initialize this stack component.

 {}

Raises:

Type Description
ValueError

If an attribute that requires custom pydantic validation is passed as a secret reference, or if the name attribute was passed as a secret reference.
Source code in src/zenml/stack/stack_component.py 
 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references don't clash with pydantic validation.

    StackComponents allow the specification of all their string attributes
    using secret references of the form `{{secret_name.key}}`. This however
    is only possible when the stack component does not perform any explicit
    validation of this attribute using pydantic validators. If this were
    the case, the validation would run on the secret reference and would
    fail or in the worst case, modify the secret reference and lead to
    unexpected behavior. This method ensures that no attributes that require
    custom pydantic validation are set as secret references.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using
            plain-text secrets.
        **kwargs: Arguments to initialize this stack component.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            is passed as a secret reference, or if the `name` attribute
            was passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}` for a `{self.__class__.__name__}` "
                    "stack component. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure your stack "
                    "components with secrets here: "
                    "https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if pydantic_utils.has_validators(
            pydantic_class=self.__class__, field_name=key
        ):
            raise ValueError(
                f"Passing the stack component attribute `{key}` as a "
                "secret reference is not allowed as additional validation "
                "is required for this attribute."
            )

    super().__init__(**kwargs)
Attributes
is_local: bool property

Checks if this stack component is running locally.

Concrete stack component configuration classes should override this method to return True if the stack component is relying on local resources or capabilities (e.g. local filesystem, local database or other services).

Examples:

  • Artifact Stores that store artifacts in the local filesystem
  • Orchestrators that are connected to local orchestration runtime services (e.g. local Kubernetes clusters, Docker containers etc).

Returns:

Type Description
bool

True if this config is for a local component, False otherwise.
is_remote: bool property

Checks if this stack component is running remotely.

Concrete stack component configuration classes should override this method to return True if the stack component is running in a remote location, and it needs to access the ZenML database.

This designation is used to determine if the stack component can be used with a local ZenML database or if it requires a remote ZenML server.

Examples:

  • Orchestrators that are running pipelines in the cloud or in a location other than the local host
  • Step Operators that are running steps in the cloud or in a location other than the local host

Returns:

Type Description
bool

True if this config is for a remote component, False otherwise.
is_valid: bool property

Checks if the stack component configurations are valid.

Concrete stack component configuration classes should override this method to return False if the stack component configurations are invalid.

Returns:

Type Description
bool

True if the stack component config is valid, False otherwise.
required_secrets: Set[secret_utils.SecretReference] property

All required secrets for this stack component.

Returns:

Type Description
Set[SecretReference]

The required secrets of this stack component.
Functions

StackValidator(required_components: Optional[AbstractSet[StackComponentType]] = None, custom_validation_function: Optional[Callable[[Stack], Tuple[bool, str]]] = None)

A StackValidator is used to validate a stack configuration.

Each StackComponent can provide a StackValidator to make sure it is compatible with all components of the stack. The KubeflowOrchestrator for example will always require the stack to have a container registry in order to push the docker images that are required to run a pipeline in Kubeflow Pipelines.

Initializes a StackValidator instance.

Parameters:

Name Type Description Default
required_components Optional[AbstractSet[StackComponentType]]

Optional set of stack components that must exist in the stack.

 None
custom_validation_function Optional[Callable[[Stack], Tuple[bool, str]]]

Optional function that returns whether a stack is valid and an error message to show if not valid.

 None
Source code in src/zenml/stack/stack_validator.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(
    self,
    required_components: Optional[AbstractSet[StackComponentType]] = None,
    custom_validation_function: Optional[
        Callable[["Stack"], Tuple[bool, str]]
    ] = None,
):
    """Initializes a `StackValidator` instance.

    Args:
        required_components: Optional set of stack components that must
            exist in the stack.
        custom_validation_function: Optional function that returns whether
            a stack is valid and an error message to show if not valid.
    """
    self._required_components = required_components or set()
    self._custom_validation_function = custom_validation_function
Functions
validate(stack: Stack) -> None

Validates the given stack.

Checks if the stack contains all the required components and passes the custom validation function of the validator.

Parameters:

Name Type Description Default
stack Stack

The stack to validate.

 required

Raises:

Type Description
StackValidationError

If the stack does not meet all the validation criteria.
Source code in src/zenml/stack/stack_validator.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def validate(self, stack: "Stack") -> None:
    """Validates the given stack.

    Checks if the stack contains all the required components and passes
    the custom validation function of the validator.

    Args:
        stack: The stack to validate.

    Raises:
        StackValidationError: If the stack does not meet all the
            validation criteria.
    """
    existing_components = set(c.type for c in stack.all_components)

    missing_components = self._required_components - existing_components
    if missing_components:
        raise StackValidationError(
            f"Missing stack components {missing_components} for "
            f"stack: {stack.name}"
        )

    if self._custom_validation_function:
        valid, err_msg = self._custom_validation_function(stack)
        if not valid:
            raise StackValidationError(
                f"Custom validation function failed to validate "
                f"stack '{stack.name}': {err_msg}"
            )

Modules

authentication_mixin

Stack component mixin for authentication.

Classes
AuthenticationConfigMixin(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: StackComponentConfig

Base config for authentication mixins.

Any stack component that implements AuthenticationMixin should have a config that inherits from this class.

Field descriptions are defined inline using Field() descriptors.

Source code in src/zenml/stack/stack_component.py 
 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references don't clash with pydantic validation.

    StackComponents allow the specification of all their string attributes
    using secret references of the form `{{secret_name.key}}`. This however
    is only possible when the stack component does not perform any explicit
    validation of this attribute using pydantic validators. If this were
    the case, the validation would run on the secret reference and would
    fail or in the worst case, modify the secret reference and lead to
    unexpected behavior. This method ensures that no attributes that require
    custom pydantic validation are set as secret references.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using
            plain-text secrets.
        **kwargs: Arguments to initialize this stack component.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            is passed as a secret reference, or if the `name` attribute
            was passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}` for a `{self.__class__.__name__}` "
                    "stack component. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure your stack "
                    "components with secrets here: "
                    "https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if pydantic_utils.has_validators(
            pydantic_class=self.__class__, field_name=key
        ):
            raise ValueError(
                f"Passing the stack component attribute `{key}` as a "
                "secret reference is not allowed as additional validation "
                "is required for this attribute."
            )

    super().__init__(**kwargs)
AuthenticationMixin(name: str, id: UUID, config: StackComponentConfig, flavor: str, type: StackComponentType, user: Optional[UUID], created: datetime, updated: datetime, environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None, labels: Optional[Dict[str, Any]] = None, connector_requirements: Optional[ServiceConnectorRequirements] = None, connector: Optional[UUID] = None, connector_resource_id: Optional[str] = None, *args: Any, **kwargs: Any)

Bases: StackComponent

Stack component mixin for authentication.

Any stack component that implements this mixin should have a config that inherits from AuthenticationConfigMixin.

Source code in src/zenml/stack/stack_component.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
def __init__(
    self,
    name: str,
    id: UUID,
    config: StackComponentConfig,
    flavor: str,
    type: StackComponentType,
    user: Optional[UUID],
    created: datetime,
    updated: datetime,
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
    labels: Optional[Dict[str, Any]] = None,
    connector_requirements: Optional[ServiceConnectorRequirements] = None,
    connector: Optional[UUID] = None,
    connector_resource_id: Optional[str] = None,
    *args: Any,
    **kwargs: Any,
):
    """Initializes a StackComponent.

    Args:
        name: The name of the component.
        id: The unique ID of the component.
        config: The config of the component.
        flavor: The flavor of the component.
        type: The type of the component.
        user: The ID of the user who created the component.
        created: The creation time of the component.
        updated: The last update time of the component.
        environment: Environment variables to set when running on this
            component.
        secrets: Secrets to set as environment variables when running on
            this component.
        labels: The labels of the component.
        connector_requirements: The requirements for the connector.
        connector: The ID of a connector linked to the component.
        connector_resource_id: The custom resource ID to access through
            the connector.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Raises:
        ValueError: If a secret reference is passed as name.
    """
    if secret_utils.is_secret_reference(name):
        raise ValueError(
            "Passing the `name` attribute of a stack component as a "
            "secret reference is not allowed."
        )

    self.id = id
    self.name = name
    self._config = config
    self.flavor = flavor
    self.type = type
    self.user = user
    self.created = created
    self.updated = updated
    self.labels = labels
    self.environment = environment or {}
    self.secrets = secrets or []
    self.connector_requirements = connector_requirements
    self.connector = connector
    self.connector_resource_id = connector_resource_id
    self._connector_instance: Optional[ServiceConnector] = None
Attributes
config: AuthenticationConfigMixin property

Returns the AuthenticationConfigMixin config.

Returns:

Type Description
AuthenticationConfigMixin

The configuration.
Functions
get_authentication_secret() -> Optional[SecretResponse]

Gets the secret referred to by the authentication secret attribute.

Returns:

Type Description
Optional[SecretResponse]

The secret if the authentication_secret attribute is set,
Optional[SecretResponse]

None otherwise.

Raises:

Type Description
KeyError

If the secret does not exist.
Source code in src/zenml/stack/authentication_mixin.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def get_authentication_secret(
    self,
) -> Optional[SecretResponse]:
    """Gets the secret referred to by the authentication secret attribute.

    Returns:
        The secret if the `authentication_secret` attribute is set,
        `None` otherwise.

    Raises:
        KeyError: If the secret does not exist.
    """
    if not self.config.authentication_secret:
        return None

    # Try to resolve the secret using the secret store
    try:
        return Client().get_secret_by_name_and_private_status(
            name=self.config.authentication_secret,
        )
    except (KeyError, NotImplementedError):
        raise KeyError(
            f"The authentication secret {self.config.authentication_secret} "
            f"referenced by the `{self.name}` `{self.type}` stack "
            "component does not exist."
        )
get_typed_authentication_secret(expected_schema_type: Type[T]) -> Optional[T]

Gets a typed secret referred to by the authentication secret attribute.

Parameters:

Name Type Description Default
expected_schema_type Type[T]

A Pydantic model class that represents the expected schema type of the secret.

 required

Returns:

Type Description
Optional[T]

The secret values extracted from the secret and converted into the
Optional[T]

indicated Pydantic type, if the authentication_secret attribute is
Optional[T]

set, None otherwise.

Raises:

Type Description
TypeError

If the secret cannot be converted into the indicated Pydantic type.
Source code in src/zenml/stack/authentication_mixin.py 
 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
112
113
114
115
116
117
118
119
def get_typed_authentication_secret(
    self, expected_schema_type: Type[T]
) -> Optional[T]:
    """Gets a typed secret referred to by the authentication secret attribute.

    Args:
        expected_schema_type: A Pydantic model class that represents the
            expected schema type of the secret.

    Returns:
        The secret values extracted from the secret and converted into the
        indicated Pydantic type, if the `authentication_secret` attribute is
        set, `None` otherwise.

    Raises:
        TypeError: If the secret cannot be converted into the indicated
            Pydantic type.
    """
    secret = self.get_authentication_secret()

    if not secret:
        return None

    try:
        typed_secret = expected_schema_type(
            **secret.secret_values,
        )
    except (TypeError, ValueError) as e:
        raise TypeError(
            f"Authentication secret `{self.config.authentication_secret}` "
            f"referenced by the `{self.name}` `{self.type}` stack component"
            f"could not be converted to {expected_schema_type}: {e}"
        )

    return typed_secret

flavor

Base ZenML Flavor implementation.

Classes
Flavor

Class for ZenML Flavors.

Attributes
config_class: Type[StackComponentConfig] abstractmethod property

Returns StackComponentConfig config class.

Returns:

Type Description
Type[StackComponentConfig]

The config class.
config_schema: Dict[str, Any] property

The config schema for a flavor.

Returns:

Type Description
Dict[str, Any]

The config schema.
display_name: Optional[str] property

The display name of the flavor.

By default, converts the technical name to a human-readable format. For example, "vm_kubernetes" becomes "VM Kubernetes". Flavors can override this to provide custom display names.

Returns:

Type Description
Optional[str]

The display name of the flavor.
docs_url: Optional[str] property

A url to point at docs explaining this flavor.

Returns:

Type Description
Optional[str]

A flavor docs url.
implementation_class: Type[StackComponent] abstractmethod property

Implementation class for this flavor.

Returns:

Type Description
Type[StackComponent]

The implementation class for this flavor.
logo_url: Optional[str] property

A url to represent the flavor in the dashboard.

Returns:

Type Description
Optional[str]

The flavor logo.
name: str abstractmethod property

The flavor name.

Returns:

Type Description
str

The flavor name.
sdk_docs_url: Optional[str] property

A url to point at SDK docs explaining this flavor.

Returns:

Type Description
Optional[str]

A flavor SDK docs url.
service_connector_requirements: Optional[ServiceConnectorRequirements] property

Service connector resource requirements for service connectors.

Specifies resource requirements that are used to filter the available service connector types that are compatible with this flavor.

Returns:

Type Description
Optional[ServiceConnectorRequirements]

Requirements for compatible service connectors, if a service
Optional[ServiceConnectorRequirements]

connector is required for this flavor.
type: StackComponentType abstractmethod property

The stack component type.

Returns:

Type Description
StackComponentType

The stack component type.
Functions
from_model(flavor_model: FlavorResponse) -> Flavor classmethod

Loads a flavor from a model.

Parameters:

Name Type Description Default
flavor_model FlavorResponse

The model to load from.

 required

Raises:

Type Description
CustomFlavorImportError

If the custom flavor can't be imported.
ImportError

If the flavor can't be imported.

Returns:

Type Description
Flavor

The loaded flavor.
Source code in src/zenml/stack/flavor.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
@classmethod
def from_model(cls, flavor_model: FlavorResponse) -> "Flavor":
    """Loads a flavor from a model.

    Args:
        flavor_model: The model to load from.

    Raises:
        CustomFlavorImportError: If the custom flavor can't be imported.
        ImportError: If the flavor can't be imported.

    Returns:
        The loaded flavor.
    """
    try:
        flavor = source_utils.load(flavor_model.source)()
    except (ModuleNotFoundError, ImportError, NotImplementedError) as err:
        if flavor_model.is_custom:
            flavor_module, _ = flavor_model.source.rsplit(".", maxsplit=1)
            expected_file_path = os.path.join(
                source_utils.get_source_root(),
                flavor_module.replace(".", os.path.sep),
            )
            raise CustomFlavorImportError(
                f"Couldn't import custom flavor {flavor_model.name}: "
                f"{err}. Make sure the custom flavor class "
                f"`{flavor_model.source}` is importable. If it is part of "
                "a library, make sure it is installed. If "
                "it is a local code file, make sure it exists at "
                f"`{expected_file_path}.py`."
            )
        else:
            raise ImportError(
                f"Couldn't import flavor {flavor_model.name}: {err}"
            )
    return cast(Flavor, flavor)
generate_default_docs_url() -> str

Generate the doc urls for all inbuilt and integration flavors.

Note that this method is not going to be useful for custom flavors, which do not have any docs in the main zenml docs.

Returns:

Type Description
str

The complete url to the zenml documentation
Source code in src/zenml/stack/flavor.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def generate_default_docs_url(self) -> str:
    """Generate the doc urls for all inbuilt and integration flavors.

    Note that this method is not going to be useful for custom flavors,
    which do not have any docs in the main zenml docs.

    Returns:
        The complete url to the zenml documentation
    """
    component_type = self.type.plural.replace("_", "-")
    name = self.name.replace("_", "-")

    base = "https://docs.zenml.io"
    return f"{base}/stack-components/{component_type}/{name}"
generate_default_sdk_docs_url() -> str

Generate SDK docs url for a flavor.

Returns:

Type Description
str

The complete url to the zenml SDK docs
Source code in src/zenml/stack/flavor.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def generate_default_sdk_docs_url(self) -> str:
    """Generate SDK docs url for a flavor.

    Returns:
        The complete url to the zenml SDK docs
    """
    from zenml import __version__

    base = f"https://sdkdocs.zenml.io/{__version__}"

    component_type = self.type.plural

    if "zenml.integrations" in self.__module__:
        # Get integration name out of module path which will look something
        #  like this "zenml.integrations.<integration>....
        integration = self.__module__.split(
            "zenml.integrations.", maxsplit=1
        )[1].split(".")[0]

        # Get the config class name to point to the specific class
        config_class_name = self.config_class.__name__

        return (
            f"{base}/integration_code_docs"
            f"/integrations-{integration}"
            f"#zenml.integrations.{integration}.flavors.{config_class_name}"
        )

    else:
        return (
            f"{base}/core_code_docs/core-{component_type}/"
            f"#{self.__module__}"
        )
to_model(integration: Optional[str] = None, is_custom: bool = True) -> FlavorRequest

Converts a flavor to a model.

Parameters:

Name Type Description Default
integration Optional[str]

The integration to use for the model.

 None
is_custom bool

Whether the flavor is a custom flavor.

 True

Returns:

Type Description
FlavorRequest

The model.
Source code in src/zenml/stack/flavor.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def to_model(
    self,
    integration: Optional[str] = None,
    is_custom: bool = True,
) -> FlavorRequest:
    """Converts a flavor to a model.

    Args:
        integration: The integration to use for the model.
        is_custom: Whether the flavor is a custom flavor.

    Returns:
        The model.
    """
    connector_requirements = self.service_connector_requirements
    connector_type = (
        connector_requirements.connector_type
        if connector_requirements
        else None
    )
    resource_type = (
        connector_requirements.resource_type
        if connector_requirements
        else None
    )
    resource_id_attr = (
        connector_requirements.resource_id_attr
        if connector_requirements
        else None
    )

    model = FlavorRequest(
        name=self.name,
        display_name=self.display_name,
        type=self.type,
        source=source_utils.resolve(self.__class__).import_path,
        config_schema=self.config_schema,
        connector_type=connector_type,
        connector_resource_type=resource_type,
        connector_resource_id_attr=resource_id_attr,
        integration=integration,
        logo_url=self.logo_url,
        docs_url=self.docs_url,
        sdk_docs_url=self.sdk_docs_url,
        is_custom=is_custom,
    )
    return model
Functions
validate_flavor_source(source: str, component_type: StackComponentType) -> Type[Flavor]

Import a StackComponent class from a given source and validate its type.

Parameters:

Name Type Description Default
source str

source path of the implementation

 required
component_type StackComponentType

the type of the stack component

 required

Returns:

Type Description
Type[Flavor]

the imported class

Raises:

Type Description
ValueError

If ZenML cannot find the given module path
TypeError

If the given module path does not point to a subclass of a StackComponent which has the right component type.
Source code in src/zenml/stack/flavor.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
def validate_flavor_source(
    source: str, component_type: StackComponentType
) -> Type["Flavor"]:
    """Import a StackComponent class from a given source and validate its type.

    Args:
        source: source path of the implementation
        component_type: the type of the stack component

    Returns:
        the imported class

    Raises:
        ValueError: If ZenML cannot find the given module path
        TypeError: If the given module path does not point to a subclass of a
            StackComponent which has the right component type.
    """
    from zenml.stack.stack_component import (
        StackComponent,
        StackComponentConfig,
    )
    from zenml.utils import source_utils

    try:
        flavor_class = source_utils.load(source)
    except (ValueError, AttributeError, ImportError) as e:
        raise ValueError(
            f"ZenML can not import the flavor class '{source}': {e}"
        )

    if not (
        isinstance(flavor_class, type) and issubclass(flavor_class, Flavor)
    ):
        raise TypeError(
            f"The source '{source}' does not point to a subclass of the ZenML"
            f"Flavor."
        )

    flavor = flavor_class()
    try:
        impl_class = flavor.implementation_class
    except (ModuleNotFoundError, ImportError, NotImplementedError) as e:
        raise ValueError(
            f"The implementation class defined within the "
            f"'{flavor_class.__name__}' can not be imported: {e}"
        )

    if not issubclass(impl_class, StackComponent):
        raise TypeError(
            f"The implementation class '{impl_class.__name__}' of a flavor "
            f"needs to be a subclass of the ZenML StackComponent."
        )

    if flavor.type != component_type:  # noqa
        raise TypeError(
            f"The source points to a {impl_class.type}, not a "  # noqa
            f"{component_type}."
        )

    try:
        conf_class = flavor.config_class
    except (ModuleNotFoundError, ImportError, NotImplementedError) as e:
        raise ValueError(
            f"The config class defined within the "
            f"'{flavor_class.__name__}' can not be imported: {e}"
        )

    if not issubclass(conf_class, StackComponentConfig):
        raise TypeError(
            f"The config class '{conf_class.__name__}' of a flavor "
            f"needs to be a subclass of the ZenML StackComponentConfig."
        )

    return flavor_class
Modules

flavor_registry

Implementation of the ZenML flavor registry.

Classes
FlavorRegistry()

Registry for stack component flavors.

The flavors defined by ZenML must be registered here.

Initialization of the flavors.

Source code in src/zenml/stack/flavor_registry.py 
40
41
42
43
44
def __init__(self) -> None:
    """Initialization of the flavors."""
    self._flavors: DefaultDict[
        StackComponentType, Dict[str, FlavorResponse]
    ] = defaultdict(dict)
Attributes
builtin_flavors: List[Type[Flavor]] property

A list of all default in-built flavors.

Returns:

Type Description
List[Type[Flavor]]

A list of builtin flavors.
integration_flavors: List[Type[Flavor]] property

A list of all default integration flavors.

Returns:

Type Description
List[Type[Flavor]]

A list of integration flavors.
Functions
register_builtin_flavors(store: BaseZenStore) -> None

Registers the default built-in flavors.

Parameters:

Name Type Description Default
store BaseZenStore

The instance of the zen_store to use

 required
Source code in src/zenml/stack/flavor_registry.py 
112
113
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
def register_builtin_flavors(self, store: BaseZenStore) -> None:
    """Registers the default built-in flavors.

    Args:
        store: The instance of the zen_store to use
    """
    with analytics_disabler():
        for flavor in self.builtin_flavors:
            flavor_request_model = flavor().to_model(
                integration="built-in",
                is_custom=False,
            )
            existing_flavor = store.list_flavors(
                FlavorFilter(
                    name=flavor_request_model.name,
                    type=flavor_request_model.type,
                )
            )

            if len(existing_flavor) == 0:
                store.create_flavor(flavor_request_model)
            else:
                flavor_update_model = FlavorUpdate.model_validate(
                    dict(flavor_request_model)
                )
                store.update_flavor(
                    existing_flavor[0].id, flavor_update_model
                )
register_flavors(store: BaseZenStore) -> None

Register all flavors to the DB.

Parameters:

Name Type Description Default
store BaseZenStore

The instance of a store to use for persistence

 required
Source code in src/zenml/stack/flavor_registry.py 
46
47
48
49
50
51
52
53
def register_flavors(self, store: BaseZenStore) -> None:
    """Register all flavors to the DB.

    Args:
        store: The instance of a store to use for persistence
    """
    self.register_builtin_flavors(store=store)
    self.register_integration_flavors(store=store)
register_integration_flavors(store: BaseZenStore) -> None staticmethod

Registers the flavors implemented by integrations.

Parameters:

Name Type Description Default
store BaseZenStore

The instance of the zen_store to use

 required
Source code in src/zenml/stack/flavor_registry.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@staticmethod
def register_integration_flavors(store: BaseZenStore) -> None:
    """Registers the flavors implemented by integrations.

    Args:
        store: The instance of the zen_store to use
    """
    with analytics_disabler():
        for name, integration in integration_registry.integrations.items():
            try:
                integrated_flavors = integration.flavors()
                for flavor in integrated_flavors:
                    flavor_request_model = flavor().to_model(
                        integration=name,
                        is_custom=False,
                    )
                    existing_flavor = store.list_flavors(
                        FlavorFilter(
                            name=flavor_request_model.name,
                            type=flavor_request_model.type,
                        )
                    )
                    if len(existing_flavor) == 0:
                        store.create_flavor(flavor_request_model)
                    else:
                        flavor_update_model = FlavorUpdate.model_validate(
                            dict(flavor_request_model)
                        )
                        store.update_flavor(
                            existing_flavor[0].id, flavor_update_model
                        )
            except Exception as e:
                logger.warning(
                    f"Integration {name} failed to register flavors. "
                    f"Error: {e}"
                )
Functions

stack

Implementation of the ZenML Stack class.

Classes
Stack(id: UUID, name: str, *, environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None, orchestrator: BaseOrchestrator, artifact_store: BaseArtifactStore, container_registry: Optional[BaseContainerRegistry] = None, step_operator: Optional[Union[BaseStepOperator, List[BaseStepOperator]]] = None, feature_store: Optional[BaseFeatureStore] = None, model_deployer: Optional[BaseModelDeployer] = None, experiment_tracker: Optional[Union[BaseExperimentTracker, List[BaseExperimentTracker]]] = None, alerter: Optional[Union[BaseAlerter, List[BaseAlerter]]] = None, annotator: Optional[BaseAnnotator] = None, data_validator: Optional[BaseDataValidator] = None, image_builder: Optional[BaseImageBuilder] = None, model_registry: Optional[BaseModelRegistry] = None, deployer: Optional[BaseDeployer] = None, log_store: Optional[BaseLogStore] = None)

ZenML stack class.

A ZenML stack is a collection of multiple stack components that are required to run ZenML pipelines. Some of these components (orchestrator, and artifact store) are required to run any kind of pipeline, other components like the container registry are only required if other stack components depend on them.

Initializes and validates a stack instance.

Parameters:

Name Type Description Default
id UUID

Unique ID of the stack.

 required
name str

Name of the stack.

 required
environment Optional[Dict[str, str]]

Environment variables to set when running on this stack.

 None
secrets Optional[List[UUID]]

Secrets to set as environment variables when running on this stack.

 None
orchestrator BaseOrchestrator

Orchestrator component of the stack.

 required
artifact_store BaseArtifactStore

Artifact store component of the stack.

 required
container_registry Optional[BaseContainerRegistry]

Container registry component of the stack.

 None
step_operator Optional[Union[BaseStepOperator, List[BaseStepOperator]]]

Step operator component of the stack.

 None
feature_store Optional[BaseFeatureStore]

Feature store component of the stack.

 None
model_deployer Optional[BaseModelDeployer]

Model deployer component of the stack.

 None
experiment_tracker Optional[Union[BaseExperimentTracker, List[BaseExperimentTracker]]]

Experiment tracker component of the stack.

 None
alerter Optional[Union[BaseAlerter, List[BaseAlerter]]]

Alerter component of the stack.

 None
annotator Optional[BaseAnnotator]

Annotator component of the stack.

 None
data_validator Optional[BaseDataValidator]

Data validator component of the stack.

 None
image_builder Optional[BaseImageBuilder]

Image builder component of the stack.

 None
model_registry Optional[BaseModelRegistry]

Model registry component of the stack.

 None
deployer Optional[BaseDeployer]

Deployer component of the stack.

 None
log_store Optional[BaseLogStore]

Log store component of the stack.

 None
Source code in src/zenml/stack/stack.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
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
153
154
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
181
182
183
184
def __init__(
    self,
    id: UUID,
    name: str,
    *,
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
    orchestrator: "BaseOrchestrator",
    artifact_store: "BaseArtifactStore",
    container_registry: Optional["BaseContainerRegistry"] = None,
    step_operator: Optional[
        Union["BaseStepOperator", List["BaseStepOperator"]]
    ] = None,
    feature_store: Optional["BaseFeatureStore"] = None,
    model_deployer: Optional["BaseModelDeployer"] = None,
    experiment_tracker: Optional[
        Union["BaseExperimentTracker", List["BaseExperimentTracker"]]
    ] = None,
    alerter: Optional[Union["BaseAlerter", List["BaseAlerter"]]] = None,
    annotator: Optional["BaseAnnotator"] = None,
    data_validator: Optional["BaseDataValidator"] = None,
    image_builder: Optional["BaseImageBuilder"] = None,
    model_registry: Optional["BaseModelRegistry"] = None,
    deployer: Optional["BaseDeployer"] = None,
    log_store: Optional["BaseLogStore"] = None,
):
    """Initializes and validates a stack instance.

    Args:
        id: Unique ID of the stack.
        name: Name of the stack.
        environment: Environment variables to set when running on this
            stack.
        secrets: Secrets to set as environment variables when running on
            this stack.
        orchestrator: Orchestrator component of the stack.
        artifact_store: Artifact store component of the stack.
        container_registry: Container registry component of the stack.
        step_operator: Step operator component of the stack.
        feature_store: Feature store component of the stack.
        model_deployer: Model deployer component of the stack.
        experiment_tracker: Experiment tracker component of the stack.
        alerter: Alerter component of the stack.
        annotator: Annotator component of the stack.
        data_validator: Data validator component of the stack.
        image_builder: Image builder component of the stack.
        model_registry: Model registry component of the stack.
        deployer: Deployer component of the stack.
        log_store: Log store component of the stack.
    """
    self._id = id
    self._name = name
    self._environment = environment or {}
    self._secrets = secrets or []

    self._components: Dict[StackComponentType, List["StackComponent"]] = (
        defaultdict(list)
    )

    for component_input in [
        orchestrator,
        artifact_store,
        container_registry,
        step_operator,
        feature_store,
        model_deployer,
        experiment_tracker,
        alerter,
        annotator,
        data_validator,
        image_builder,
        model_registry,
        deployer,
        log_store,
    ]:
        components: List["StackComponent"] = list()
        if component_input:
            if isinstance(component_input, list):
                components.extend(component_input)
            else:
                components.append(component_input)

            for component in components:
                self._components[component.type].append(component)
Attributes
alerter: Optional[BaseAlerter] property

The alerter of the stack.

Returns:

Type Description
Optional[BaseAlerter]

The alerter of the stack.
alerters: Dict[str, BaseAlerter] property

All attached alerters keyed by component name.

Returns:

Type Description
Dict[str, BaseAlerter]

The stack alerters keyed by their names.
all_components: List[StackComponent] property

All attached components of the stack.

Returns:

Type Description
List[StackComponent]

A flat list of all components attached to the stack.
annotator: Optional[BaseAnnotator] property

The annotator of the stack.

Returns:

Type Description
Optional[BaseAnnotator]

The annotator of the stack.
apt_packages: List[str] property

List of APT package requirements for the stack.

Returns:

Type Description
List[str]

A list of APT package requirements for the stack.
artifact_store: BaseArtifactStore property

The artifact store of the stack.

Returns:

Type Description
BaseArtifactStore

The artifact store of the stack.
component_flavor_metadata: Dict[str, str] property

Flavor metadata for attached components grouped by type.

Returns:

Type Description
Dict[str, str]

A mapping from stack component type values to comma-separated
Dict[str, str]

flavor names. Repeated flavors are only included once and the
Dict[str, str]

component order is preserved.
components: Dict[StackComponentType, StackComponent] property

All components of the stack.

Returns:

Type Description
Dict[StackComponentType, StackComponent]

A dictionary of all components of the stack.
container_registry: Optional[BaseContainerRegistry] property

The container registry of the stack.

Returns:

Type Description
Optional[BaseContainerRegistry]

The container registry of the stack or None if the stack does not
Optional[BaseContainerRegistry]

have a container registry.
data_validator: Optional[BaseDataValidator] property

The data validator of the stack.

Returns:

Type Description
Optional[BaseDataValidator]

The data validator of the stack.
deployer: Optional[BaseDeployer] property

The deployer of the stack.

Returns:

Type Description
Optional[BaseDeployer]

The deployer of the stack.
environment: Dict[str, str] property

Environment variables to set when running on this stack.

Returns:

Type Description
Dict[str, str]

Environment variables to set when running on this stack.
experiment_tracker: Optional[BaseExperimentTracker] property

The experiment tracker of the stack.

Returns:

Type Description
Optional[BaseExperimentTracker]

The experiment tracker of the stack.
experiment_trackers: Dict[str, BaseExperimentTracker] property

All attached experiment trackers keyed by component name.

Returns:

Type Description
Dict[str, BaseExperimentTracker]

The stack experiment trackers keyed by their names.
feature_store: Optional[BaseFeatureStore] property

The feature store of the stack.

Returns:

Type Description
Optional[BaseFeatureStore]

The feature store of the stack.
id: UUID property

The ID of the stack.

Returns:

Type Description
UUID

The ID of the stack.
image_builder: Optional[BaseImageBuilder] property

The image builder of the stack.

Returns:

Type Description
Optional[BaseImageBuilder]

The image builder of the stack.
log_store: BaseLogStore property

The log store of the stack.

Returns:

Type Description
BaseLogStore

The log store of the stack.
model_deployer: Optional[BaseModelDeployer] property

The model deployer of the stack.

Returns:

Type Description
Optional[BaseModelDeployer]

The model deployer of the stack.
model_registry: Optional[BaseModelRegistry] property

The model registry of the stack.

Returns:

Type Description
Optional[BaseModelRegistry]

The model registry of the stack.
name: str property

The name of the stack.

Returns:

Name Type Description
str str

The name of the stack.
orchestrator: BaseOrchestrator property

The orchestrator of the stack.

Returns:

Type Description
BaseOrchestrator

The orchestrator of the stack.
required_secrets: Set[secret_utils.SecretReference] property

All required secrets for this stack.

Returns:

Type Description
Set[SecretReference]

The required secrets of this stack.
requires_remote_server: bool property

If the stack requires a remote ZenServer to run.

This is the case if any code is getting executed remotely. This is the case for both remote orchestrators as well as remote step operators.

Returns:

Type Description
bool

If the stack requires a remote ZenServer to run.
secrets: List[UUID] property

Secrets to set as environment variables when running on this stack.

Returns:

Type Description
List[UUID]

Secrets to set as environment variables when running on this stack.
setting_classes: Dict[str, Type[BaseSettings]] property

Setting classes of all components of this stack.

Returns:

Type Description
Dict[str, Type[BaseSettings]]

All setting classes and their respective keys.
step_operator: Optional[BaseStepOperator] property

The step operator of the stack.

Returns:

Type Description
Optional[BaseStepOperator]

The step operator of the stack.
step_operators: Dict[str, BaseStepOperator] property

All attached step operators keyed by component name.

Returns:

Type Description
Dict[str, BaseStepOperator]

The stack step operators keyed by their names.
Functions
check_local_paths() -> bool

Checks if the stack has local paths.

Returns:

Type Description
bool

True if the stack has local paths, False otherwise.

Raises:

Type Description
ValueError

If the stack has local paths that do not conform to the convention that all local path must be relative to the local stores directory.
Source code in src/zenml/stack/stack.py 
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
def check_local_paths(self) -> bool:
    """Checks if the stack has local paths.

    Returns:
        True if the stack has local paths, False otherwise.

    Raises:
        ValueError: If the stack has local paths that do not conform to
            the convention that all local path must be relative to the
            local stores directory.
    """
    from zenml.config.global_config import GlobalConfiguration

    local_stores_path = GlobalConfiguration().local_stores_path

    # go through all stack components and identify those that advertise
    # a local path where they persist information that they need to be
    # available when running pipelines.
    has_local_paths = False
    for stack_comp in self.all_components:
        local_path = stack_comp.local_path
        if not local_path:
            continue
        # double-check this convention, just in case it wasn't respected
        # as documented in `StackComponent.local_path`
        if not local_path.startswith(local_stores_path):
            raise ValueError(
                f"Local path {local_path} for component "
                f"{stack_comp.name} is not in the local stores "
                f"directory ({local_stores_path})."
            )
        has_local_paths = True

    return has_local_paths
cleanup_step_run(info: StepRunInfo, step_failed: bool) -> None

Cleans up resources after the step run is finished.

Parameters:

Name Type Description Default
info StepRunInfo

Info about the step that was executed.

 required
step_failed bool

Whether the step failed.

 required
Source code in src/zenml/stack/stack.py 
1553
1554
1555
1556
1557
1558
1559
1560
1561
def cleanup_step_run(self, info: "StepRunInfo", step_failed: bool) -> None:
    """Cleans up resources after the step run is finished.

    Args:
        info: Info about the step that was executed.
        step_failed: Whether the step failed.
    """
    for component in self._get_active_components_for_step(info.config):
        component.cleanup_step_run(info=info, step_failed=step_failed)
deploy_pipeline(snapshot: PipelineSnapshotResponse, deployment_name: str, timeout: Optional[int] = None) -> DeploymentResponse

Deploys a pipeline on this stack.

Parameters:

Name Type Description Default
snapshot PipelineSnapshotResponse

The pipeline snapshot.

 required
deployment_name str

The name to use for the deployment.

 required
timeout Optional[int]

The maximum time in seconds to wait for the pipeline to be deployed.

 None

Returns:

Type Description
DeploymentResponse

The deployment response.

Raises:

Type Description
RuntimeError

If the stack does not have a deployer.
Source code in src/zenml/stack/stack.py 
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
def deploy_pipeline(
    self,
    snapshot: "PipelineSnapshotResponse",
    deployment_name: str,
    timeout: Optional[int] = None,
) -> "DeploymentResponse":
    """Deploys a pipeline on this stack.

    Args:
        snapshot: The pipeline snapshot.
        deployment_name: The name to use for the deployment.
        timeout: The maximum time in seconds to wait for the pipeline to be
            deployed.

    Returns:
        The deployment response.

    Raises:
        RuntimeError: If the stack does not have a deployer.
    """
    if not self.deployer:
        raise RuntimeError(
            "The stack does not have a deployer. Please add a "
            "deployer to the stack in order to deploy a pipeline."
        )

    return self.deployer.provision_deployment(
        snapshot=snapshot,
        stack=self,
        deployment_name_or_id=deployment_name,
        timeout=timeout,
    )
dict() -> Dict[str, str]

Converts the stack into a dictionary.

Returns:

Type Description
Dict[str, str]

A dictionary containing the stack components.
Source code in src/zenml/stack/stack.py 
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
def dict(self) -> Dict[str, str]:
    """Converts the stack into a dictionary.

    Returns:
        A dictionary containing the stack components.
    """
    component_dict = {
        component_type.value: json.dumps(
            [
                component.config.model_dump(mode="json")
                for component in components
            ],
            sort_keys=True,
        )
        for component_type, components in self._components.items()
    }
    component_dict.update({"name": self.name})
    return component_dict
from_components(id: UUID, name: str, components: Dict[StackComponentType, StackComponent], environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None) -> Stack classmethod

Creates a stack instance from a dict of stack components.

Parameters:

Name Type Description Default
id UUID

Unique ID of the stack.

 required
name str

The name of the stack.

 required
components Dict[StackComponentType, StackComponent]

The components of the stack.

 required
environment Optional[Dict[str, str]]

Environment variables to set when running on this stack.

 None
secrets Optional[List[UUID]]

Secrets to set as environment variables when running on this stack.

 None

Returns:

Type Description
Stack

A stack instance consisting of the given components.

Raises:

Type Description
TypeError

If a required component is missing or a component doesn't inherit from the expected base class.
Source code in src/zenml/stack/stack.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
@classmethod
def from_components(
    cls,
    id: UUID,
    name: str,
    components: Dict[StackComponentType, "StackComponent"],
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
) -> "Stack":
    """Creates a stack instance from a dict of stack components.

    Args:
        id: Unique ID of the stack.
        name: The name of the stack.
        components: The components of the stack.
        environment: Environment variables to set when running on this
            stack.
        secrets: Secrets to set as environment variables when running on
            this stack.

    Returns:
        A stack instance consisting of the given components.

    Raises:
        TypeError: If a required component is missing or a component
            doesn't inherit from the expected base class.
    """  # noqa: DOC502
    logger.warning(
        "The class method `Stack.from_components(...)` has been deprecated. "
        "Please use `Stack.from_components_v2(...)` instead."
    )

    from zenml.alerter import BaseAlerter
    from zenml.annotators import BaseAnnotator
    from zenml.artifact_stores import BaseArtifactStore
    from zenml.container_registries import BaseContainerRegistry
    from zenml.data_validators import BaseDataValidator
    from zenml.deployers import BaseDeployer
    from zenml.experiment_trackers import BaseExperimentTracker
    from zenml.feature_stores import BaseFeatureStore
    from zenml.image_builders import BaseImageBuilder
    from zenml.log_stores import BaseLogStore
    from zenml.model_deployers import BaseModelDeployer
    from zenml.model_registries import BaseModelRegistry
    from zenml.orchestrators import BaseOrchestrator
    from zenml.step_operators import BaseStepOperator

    def _raise_type_error(
        component: Optional["StackComponent"], expected_class: Type[Any]
    ) -> NoReturn:
        """Raises a TypeError that the component has an unexpected type.

        Args:
            component: The component that has an unexpected type.
            expected_class: The expected type of the component.

        Raises:
            TypeError: If the component has an unexpected type.
        """
        raise TypeError(
            f"Unable to create stack: Wrong stack component type "
            f"`{component.__class__.__name__}` (expected: subclass "
            f"of `{expected_class.__name__}`)"
        )

    orchestrator = components.get(StackComponentType.ORCHESTRATOR)
    if not isinstance(orchestrator, BaseOrchestrator):
        _raise_type_error(orchestrator, BaseOrchestrator)

    artifact_store = components.get(StackComponentType.ARTIFACT_STORE)
    if not isinstance(artifact_store, BaseArtifactStore):
        _raise_type_error(artifact_store, BaseArtifactStore)

    container_registry = components.get(
        StackComponentType.CONTAINER_REGISTRY
    )
    if container_registry is not None and not isinstance(
        container_registry, BaseContainerRegistry
    ):
        _raise_type_error(container_registry, BaseContainerRegistry)

    step_operator = components.get(StackComponentType.STEP_OPERATOR)
    if step_operator is not None and not isinstance(
        step_operator, BaseStepOperator
    ):
        _raise_type_error(step_operator, BaseStepOperator)

    feature_store = components.get(StackComponentType.FEATURE_STORE)
    if feature_store is not None and not isinstance(
        feature_store, BaseFeatureStore
    ):
        _raise_type_error(feature_store, BaseFeatureStore)

    model_deployer = components.get(StackComponentType.MODEL_DEPLOYER)
    if model_deployer is not None and not isinstance(
        model_deployer, BaseModelDeployer
    ):
        _raise_type_error(model_deployer, BaseModelDeployer)

    experiment_tracker = components.get(
        StackComponentType.EXPERIMENT_TRACKER
    )
    if experiment_tracker is not None and not isinstance(
        experiment_tracker, BaseExperimentTracker
    ):
        _raise_type_error(experiment_tracker, BaseExperimentTracker)

    alerter = components.get(StackComponentType.ALERTER)
    if alerter is not None and not isinstance(alerter, BaseAlerter):
        _raise_type_error(alerter, BaseAlerter)

    annotator = components.get(StackComponentType.ANNOTATOR)
    if annotator is not None and not isinstance(annotator, BaseAnnotator):
        _raise_type_error(annotator, BaseAnnotator)

    data_validator = components.get(StackComponentType.DATA_VALIDATOR)
    if data_validator is not None and not isinstance(
        data_validator, BaseDataValidator
    ):
        _raise_type_error(data_validator, BaseDataValidator)

    image_builder = components.get(StackComponentType.IMAGE_BUILDER)
    if image_builder is not None and not isinstance(
        image_builder, BaseImageBuilder
    ):
        _raise_type_error(image_builder, BaseImageBuilder)

    model_registry = components.get(StackComponentType.MODEL_REGISTRY)
    if model_registry is not None and not isinstance(
        model_registry, BaseModelRegistry
    ):
        _raise_type_error(model_registry, BaseModelRegistry)

    deployer = components.get(StackComponentType.DEPLOYER)
    if deployer is not None and not isinstance(deployer, BaseDeployer):
        _raise_type_error(deployer, BaseDeployer)

    log_store = components.get(StackComponentType.LOG_STORE)
    if log_store is not None and not isinstance(log_store, BaseLogStore):
        _raise_type_error(log_store, BaseLogStore)

    return Stack(
        id=id,
        name=name,
        environment=environment,
        secrets=secrets,
        orchestrator=orchestrator,
        artifact_store=artifact_store,
        container_registry=container_registry,
        step_operator=step_operator,
        feature_store=feature_store,
        model_deployer=model_deployer,
        experiment_tracker=experiment_tracker,
        alerter=alerter,
        annotator=annotator,
        data_validator=data_validator,
        image_builder=image_builder,
        model_registry=model_registry,
        deployer=deployer,
        log_store=log_store,
    )
from_components_v2(id: UUID, name: str, components: Mapping[StackComponentType, List[StackComponent]], environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None) -> Stack classmethod

Creates a stack instance from a dict of stack components.

Parameters:

Name Type Description Default
id UUID

Unique ID of the stack.

 required
name str

The name of the stack.

 required
components Mapping[StackComponentType, List[StackComponent]]

The components of the stack grouped by type. Callers may pass either a single component per type or a list. The first component in a repeatable component list is treated as the default component for that type.

 required
environment Optional[Dict[str, str]]

Environment variables to set when running on this stack.

 None
secrets Optional[List[UUID]]

Secrets to set as environment variables when running on this stack.

 None

Returns:

Type Description
Stack

A stack instance consisting of the given components.

Raises:

Type Description
TypeError

If a required component is missing or a component doesn't inherit from the expected base class.
Source code in src/zenml/stack/stack.py 
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
@classmethod
def from_components_v2(
    cls,
    id: UUID,
    name: str,
    components: Mapping[StackComponentType, List["StackComponent"]],
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
) -> "Stack":
    """Creates a stack instance from a dict of stack components.

    Args:
        id: Unique ID of the stack.
        name: The name of the stack.
        components: The components of the stack grouped by type. Callers may
            pass either a single component per type or a list. The first
            component in a repeatable component list is treated as the
            default component for that type.
        environment: Environment variables to set when running on this
            stack.
        secrets: Secrets to set as environment variables when running on
            this stack.

    Returns:
        A stack instance consisting of the given components.

    Raises:
        TypeError: If a required component is missing or a component
            doesn't inherit from the expected base class.
    """  # noqa: DOC502
    from zenml.alerter import BaseAlerter
    from zenml.annotators import BaseAnnotator
    from zenml.artifact_stores import BaseArtifactStore
    from zenml.container_registries import BaseContainerRegistry
    from zenml.data_validators import BaseDataValidator
    from zenml.deployers import BaseDeployer
    from zenml.experiment_trackers import BaseExperimentTracker
    from zenml.feature_stores import BaseFeatureStore
    from zenml.image_builders import BaseImageBuilder
    from zenml.log_stores import BaseLogStore
    from zenml.model_deployers import BaseModelDeployer
    from zenml.model_registries import BaseModelRegistry
    from zenml.orchestrators import BaseOrchestrator
    from zenml.step_operators import BaseStepOperator

    def _raise_type_error(
        component: Optional["StackComponent"], expected_class: Type[Any]
    ) -> NoReturn:
        """Raises a TypeError that the component has an unexpected type.

        Args:
            component: The component that has an unexpected type.
            expected_class: The expected type of the component.

        Raises:
            TypeError: If the component has an unexpected type.
        """
        raise TypeError(
            f"Unable to create stack: Wrong stack component type "
            f"`{component.__class__.__name__}` (expected: subclass "
            f"of `{expected_class.__name__}`)"
        )

    # Orchestrator
    orchestrator = components.get(StackComponentType.ORCHESTRATOR)
    if not orchestrator:
        raise TypeError("No orchestrator found in the stack.")
    if len(orchestrator) > 1:
        raise TypeError("Multiple orchestrators are not supported.")

    if not isinstance(orchestrator[0], BaseOrchestrator):
        _raise_type_error(orchestrator[0], BaseOrchestrator)

    # Artifact Store
    artifact_store = components.get(StackComponentType.ARTIFACT_STORE)
    if not artifact_store:
        raise TypeError("No artifact store found in the stack.")
    if len(artifact_store) > 1:
        raise TypeError("Multiple artifact stores are not supported.")

    if not isinstance(artifact_store[0], BaseArtifactStore):
        _raise_type_error(artifact_store[0], BaseArtifactStore)

    # Container Registry
    container_registry = components.get(
        StackComponentType.CONTAINER_REGISTRY
    )
    if container_registry:
        if len(container_registry) > 1:
            raise TypeError(
                "Multiple container registries are not supported."
            )
        if not isinstance(container_registry[0], BaseContainerRegistry):
            _raise_type_error(container_registry[0], BaseContainerRegistry)

    # Step Operator
    step_operator = components.get(StackComponentType.STEP_OPERATOR)
    if step_operator is not None and not isinstance(step_operator, list):
        step_operator = [step_operator]
    for so in step_operator or []:
        if not isinstance(so, BaseStepOperator):
            _raise_type_error(so, BaseStepOperator)

    # Feature Store
    feature_store = components.get(StackComponentType.FEATURE_STORE)
    if feature_store:
        if len(feature_store) > 1:
            raise TypeError("Multiple feature stores are not supported.")

        if not isinstance(feature_store[0], BaseFeatureStore):
            _raise_type_error(feature_store[0], BaseFeatureStore)

    # Model Deployer
    model_deployer = components.get(StackComponentType.MODEL_DEPLOYER)
    if model_deployer:
        if len(model_deployer) > 1:
            raise TypeError("Multiple model deployers are not supported.")
        if not isinstance(model_deployer[0], BaseModelDeployer):
            _raise_type_error(model_deployer[0], BaseModelDeployer)

    # Experiment Tracker
    experiment_tracker = components.get(
        StackComponentType.EXPERIMENT_TRACKER
    )
    if experiment_tracker is not None and not isinstance(
        experiment_tracker, list
    ):
        experiment_tracker = [experiment_tracker]

    for et in experiment_tracker or []:
        if not isinstance(et, BaseExperimentTracker):
            _raise_type_error(et, BaseExperimentTracker)

    # Alerter
    alerter = components.get(StackComponentType.ALERTER)
    if alerter is not None and not isinstance(alerter, list):
        alerter = [alerter]
    for a in alerter or []:
        if not isinstance(a, BaseAlerter):
            _raise_type_error(a, BaseAlerter)

    # Annotator
    annotator = components.get(StackComponentType.ANNOTATOR)
    if annotator:
        if len(annotator) > 1:
            raise TypeError("Multiple annotators are not supported.")
        if not isinstance(annotator[0], BaseAnnotator):
            _raise_type_error(annotator[0], BaseAnnotator)

    # Data Validator
    data_validator = components.get(StackComponentType.DATA_VALIDATOR)
    if data_validator:
        if len(data_validator) > 1:
            raise TypeError("Multiple data validators are not supported.")
        if not isinstance(data_validator[0], BaseDataValidator):
            _raise_type_error(data_validator[0], BaseDataValidator)

    # Image Builder
    image_builder = components.get(StackComponentType.IMAGE_BUILDER)
    if image_builder:
        if len(image_builder) > 1:
            raise TypeError("Multiple image builders are not supported.")

        if not isinstance(image_builder[0], BaseImageBuilder):
            _raise_type_error(image_builder[0], BaseImageBuilder)

    # Model Registry
    model_registry = components.get(StackComponentType.MODEL_REGISTRY)
    if model_registry:
        if len(model_registry) > 1:
            raise TypeError("Multiple model registries are not supported.")
        if not isinstance(model_registry[0], BaseModelRegistry):
            _raise_type_error(model_registry[0], BaseModelRegistry)

    # Deployer
    deployer = components.get(StackComponentType.DEPLOYER)
    if deployer:
        if len(deployer) > 1:
            raise TypeError("Multiple deployers are not supported.")
        if not isinstance(deployer[0], BaseDeployer):
            _raise_type_error(deployer[0], BaseDeployer)

    # Log Store
    log_store = components.get(StackComponentType.LOG_STORE)
    if log_store:
        if len(log_store) > 1:
            raise TypeError("Multiple log stores are not supported.")
        if not isinstance(log_store[0], BaseLogStore):
            _raise_type_error(log_store[0], BaseLogStore)

    return Stack(
        id=id,
        name=name,
        environment=environment,
        secrets=secrets,
        orchestrator=orchestrator[0],
        artifact_store=artifact_store[0],
        container_registry=cast(
            Optional["BaseContainerRegistry"],
            container_registry[0] if container_registry else None,
        ),
        step_operator=cast(
            Optional[Union["BaseStepOperator", List["BaseStepOperator"]]],
            step_operator,
        ),
        feature_store=cast(
            Optional["BaseFeatureStore"],
            feature_store[0] if feature_store else None,
        ),
        model_deployer=cast(
            Optional["BaseModelDeployer"],
            model_deployer[0] if model_deployer else None,
        ),
        experiment_tracker=cast(
            Optional[
                Union[
                    "BaseExperimentTracker",
                    List["BaseExperimentTracker"],
                ]
            ],
            experiment_tracker,
        ),
        alerter=cast(
            Optional[Union["BaseAlerter", List["BaseAlerter"]]],
            alerter,
        ),
        annotator=cast(
            Optional["BaseAnnotator"], annotator[0] if annotator else None
        ),
        data_validator=cast(
            Optional["BaseDataValidator"],
            data_validator[0] if data_validator else None,
        ),
        image_builder=cast(
            Optional["BaseImageBuilder"],
            image_builder[0] if image_builder else None,
        ),
        model_registry=cast(
            Optional["BaseModelRegistry"],
            model_registry[0] if model_registry else None,
        ),
        deployer=cast(
            Optional["BaseDeployer"], deployer[0] if deployer else None
        ),
        log_store=cast(
            Optional["BaseLogStore"], log_store[0] if log_store else None
        ),
    )
from_model(stack_model: StackResponse) -> Stack classmethod

Creates a Stack instance from a StackModel.

Parameters:

Name Type Description Default
stack_model StackResponse

The StackModel to create the Stack from.

 required

Returns:

Type Description
Stack

The created Stack instance.

Raises:

Type Description
ImportError

If a stack component's dependencies cannot be imported.
Source code in src/zenml/stack/stack.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
@classmethod
def from_model(cls, stack_model: "StackResponse") -> "Stack":
    """Creates a Stack instance from a StackModel.

    Args:
        stack_model: The StackModel to create the Stack from.

    Returns:
        The created Stack instance.

    Raises:
        ImportError: If a stack component's dependencies cannot be imported.
    """
    global _STACK_CACHE
    key = (stack_model.id, stack_model.updated)
    if key in _STACK_CACHE:
        return _STACK_CACHE[key]

    from zenml.stack import StackComponent

    # Run a hydrated list call once to avoid one request per component
    hydrated_component_models = pagination_utils.depaginate(
        Client().list_stack_components,
        stack_id=stack_model.id,
        hydrate=True,
    )

    try:
        stack_components: Dict[
            StackComponentType, List["StackComponent"]
        ] = defaultdict(list)

        for component in hydrated_component_models:
            stack_components[component.type].append(
                StackComponent.from_model(component)
            )
    except ImportError as e:
        stack_hint = (
            "\n\nTo install all requirements for this stack at "
            "once, run:\n\n"
            f"  zenml stack export-requirements "
            f"'{stack_model.name}' -o stack-requirements.txt\n"
            "  pip install -r stack-requirements.txt"
        )
        e.args = (e.args[0] + stack_hint,) + e.args[1:]
        raise

    stack = Stack.from_components_v2(
        id=stack_model.id,
        name=stack_model.name,
        environment=stack_model.environment,
        secrets=stack_model.secrets,
        components=stack_components,
    )
    _STACK_CACHE[key] = stack

    client = Client()
    if stack_model.id == client.active_stack_model.id:
        if stack_model.updated > client.active_stack_model.updated:
            if client._config:
                client._config.set_active_stack(stack_model)
            else:
                GlobalConfiguration().set_active_stack(stack_model)

    return stack
get_components_by_type(component_type: StackComponentType) -> Sequence[StackComponent]

Returns all components of a given type.

Parameters:

Name Type Description Default
component_type StackComponentType

The component type to fetch.

 required

Returns:

Type Description
Sequence[StackComponent]

A list of attached components of the given type.
Source code in src/zenml/stack/stack.py 
694
695
696
697
698
699
700
701
702
703
704
705
def get_components_by_type(
    self, component_type: StackComponentType
) -> Sequence["StackComponent"]:
    """Returns all components of a given type.

    Args:
        component_type: The component type to fetch.

    Returns:
        A list of attached components of the given type.
    """
    return self._components.get(component_type, [])
get_docker_builds(snapshot: PipelineSnapshotBase) -> List[BuildConfiguration]

Gets the Docker builds required for the stack.

Parameters:

Name Type Description Default
snapshot PipelineSnapshotBase

The pipeline snapshot for which to get the builds.

 required

Returns:

Type Description
List[BuildConfiguration]

The required Docker builds.
Source code in src/zenml/stack/stack.py 
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
def get_docker_builds(
    self, snapshot: "PipelineSnapshotBase"
) -> List["BuildConfiguration"]:
    """Gets the Docker builds required for the stack.

    Args:
        snapshot: The pipeline snapshot for which to get the builds.

    Returns:
        The required Docker builds.
    """
    return list(
        itertools.chain.from_iterable(
            component.get_docker_builds(snapshot=snapshot)
            for component in self.all_components
        )
    )
get_pipeline_run_metadata(run_id: UUID) -> Dict[UUID, Dict[str, MetadataType]]

Get general component-specific metadata for a pipeline run.

Parameters:

Name Type Description Default
run_id UUID

ID of the pipeline run.

 required

Returns:

Type Description
Dict[UUID, Dict[str, MetadataType]]

A dictionary mapping component IDs to the metadata they created.
Source code in src/zenml/stack/stack.py 
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
def get_pipeline_run_metadata(
    self, run_id: UUID
) -> Dict[UUID, Dict[str, MetadataType]]:
    """Get general component-specific metadata for a pipeline run.

    Args:
        run_id: ID of the pipeline run.

    Returns:
        A dictionary mapping component IDs to the metadata they created.
    """
    pipeline_run_metadata: Dict[UUID, Dict[str, MetadataType]] = {}
    for component in self.all_components:
        try:
            component_metadata = component.get_pipeline_run_metadata(
                run_id=run_id
            )
            if component_metadata:
                pipeline_run_metadata[component.id] = component_metadata
        except Exception as e:
            logger.warning(
                f"Extracting pipeline run metadata failed for component "
                f"'{component.name}' of type '{component.type}': {e}"
            )
    return pipeline_run_metadata
get_step_run_metadata(info: StepRunInfo) -> Dict[UUID, Dict[str, MetadataType]]

Get component-specific metadata for a step run.

Parameters:

Name Type Description Default
info StepRunInfo

Info about the step that was executed.

 required

Returns:

Type Description
Dict[UUID, Dict[str, MetadataType]]

A dictionary mapping component IDs to the metadata they created.
Source code in src/zenml/stack/stack.py 
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
def get_step_run_metadata(
    self, info: "StepRunInfo"
) -> Dict[UUID, Dict[str, MetadataType]]:
    """Get component-specific metadata for a step run.

    Args:
        info: Info about the step that was executed.

    Returns:
        A dictionary mapping component IDs to the metadata they created.
    """
    step_run_metadata: Dict[UUID, Dict[str, MetadataType]] = {}
    for component in self._get_active_components_for_step(info.config):
        try:
            component_metadata = component.get_step_run_metadata(info=info)
            if component_metadata:
                step_run_metadata[component.id] = component_metadata
        except Exception as e:
            logger.warning(
                f"Extracting step run metadata failed for component "
                f"'{component.name}' of type '{component.type}': {e}"
            )
    return step_run_metadata
prepare_pipeline_submission(snapshot: PipelineSnapshotResponse) -> None

Prepares the stack for a pipeline submission.

This method is called before a pipeline is submitted.

Parameters:

Name Type Description Default
snapshot PipelineSnapshotResponse

The pipeline snapshot

 required

Raises:

Type Description
RuntimeError

If trying to submit a pipeline that requires a remote ZenML server with a local one.
Source code in src/zenml/stack/stack.py 
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
def prepare_pipeline_submission(
    self, snapshot: "PipelineSnapshotResponse"
) -> None:
    """Prepares the stack for a pipeline submission.

    This method is called before a pipeline is submitted.

    Args:
        snapshot: The pipeline snapshot

    Raises:
        RuntimeError: If trying to submit a pipeline that requires a remote
            ZenML server with a local one.
    """
    self.validate(fail_if_secrets_missing=True)

    if self.requires_remote_server and Client().zen_store.is_local_store():
        raise RuntimeError(
            "Stacks with remote components such as remote orchestrators "
            "and step operators require a remote "
            "ZenML server. To run a pipeline with this stack you need to "
            "connect to a remote ZenML server first. Check out "
            "https://docs.zenml.io/getting-started/deploying-zenml "
            "for more information on how to deploy ZenML."
        )
prepare_step_run(info: StepRunInfo) -> None

Prepares running a step.

Parameters:

Name Type Description Default
info StepRunInfo

Info about the step that will be executed.

 required
Source code in src/zenml/stack/stack.py 
1494
1495
1496
1497
1498
1499
1500
1501
def prepare_step_run(self, info: "StepRunInfo") -> None:
    """Prepares running a step.

    Args:
        info: Info about the step that will be executed.
    """
    for component in self._get_active_components_for_step(info.config):
        component.prepare_step_run(info=info)
requirements(exclude_components: Optional[AbstractSet[StackComponentType]] = None) -> Set[str]

Set of PyPI requirements for the stack.

This method combines the requirements of all stack components (except the ones specified in exclude_components).

Parameters:

Name Type Description Default
exclude_components Optional[AbstractSet[StackComponentType]]

Set of component types for which the requirements should not be included in the output.

 None

Returns:

Type Description
Set[str]

Set of PyPI requirements.
Source code in src/zenml/stack/stack.py 
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
def requirements(
    self,
    exclude_components: Optional[AbstractSet[StackComponentType]] = None,
) -> Set[str]:
    """Set of PyPI requirements for the stack.

    This method combines the requirements of all stack components (except
    the ones specified in `exclude_components`).

    Args:
        exclude_components: Set of component types for which the
            requirements should not be included in the output.

    Returns:
        Set of PyPI requirements.
    """
    exclude_components = exclude_components or set()
    requirements = [
        component.requirements
        for component in self.all_components
        if component.type not in exclude_components
    ]
    return set.union(*requirements) if requirements else set()
submit_pipeline(snapshot: PipelineSnapshotResponse, placeholder_run: Optional[PipelineRunResponse] = None) -> None

Submits a pipeline on this stack.

Parameters:

Name Type Description Default
snapshot PipelineSnapshotResponse

The pipeline snapshot.

 required
placeholder_run Optional[PipelineRunResponse]

An optional placeholder run for the snapshot.

 None
Source code in src/zenml/stack/stack.py 
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
def submit_pipeline(
    self,
    snapshot: "PipelineSnapshotResponse",
    placeholder_run: Optional["PipelineRunResponse"] = None,
) -> None:
    """Submits a pipeline on this stack.

    Args:
        snapshot: The pipeline snapshot.
        placeholder_run: An optional placeholder run for the snapshot.
    """
    self.orchestrator.run(
        snapshot=snapshot, stack=self, placeholder_run=placeholder_run
    )
validate(fail_if_secrets_missing: bool = False) -> None

Checks whether the stack configuration is valid.

To check if a stack configuration is valid, the following criteria must be met: - the stack must have an image builder if other components require it - the StackValidator of each stack component has to validate the stack to make sure all the components are compatible with each other - the required secrets of all components need to exist

Parameters:

Name Type Description Default
fail_if_secrets_missing bool

If this is True, an error will be raised if a secret for a component is missing. Otherwise, only a warning will be logged.

 False
Source code in src/zenml/stack/stack.py 
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
def validate(
    self,
    fail_if_secrets_missing: bool = False,
) -> None:
    """Checks whether the stack configuration is valid.

    To check if a stack configuration is valid, the following criteria must
    be met:
    - the stack must have an image builder if other components require it
    - the `StackValidator` of each stack component has to validate the
        stack to make sure all the components are compatible with each other
    - the required secrets of all components need to exist

    Args:
        fail_if_secrets_missing: If this is `True`, an error will be raised
            if a secret for a component is missing. Otherwise, only a
            warning will be logged.
    """
    if handle_bool_env_var(ENV_ZENML_SKIP_STACK_VALIDATION, default=False):
        logger.debug("Skipping stack validation.")
        return

    self.validate_image_builder()
    self.validate_log_store()
    for component in self.all_components:
        if component.validator:
            component.validator.validate(stack=self)

    self._validate_secrets(raise_exception=fail_if_secrets_missing)
validate_image_builder() -> None

Validates that the stack has an image builder if required.

If the stack requires an image builder, but none is specified, a local image builder will be created and assigned to the stack to ensure backwards compatibility.

Source code in src/zenml/stack/stack.py 
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
def validate_image_builder(self) -> None:
    """Validates that the stack has an image builder if required.

    If the stack requires an image builder, but none is specified, a
    local image builder will be created and assigned to the stack to
    ensure backwards compatibility.
    """
    requires_image_builder = (
        self.orchestrator.flavor != "local"
        or self.step_operator
        or self.deployer
        or (self.model_deployer and self.model_deployer.flavor != "mlflow")
    )
    skip_default_image_builder = handle_bool_env_var(
        ENV_ZENML_SKIP_IMAGE_BUILDER_DEFAULT, default=False
    )
    if (
        requires_image_builder
        and not skip_default_image_builder
        and not self.image_builder
    ):
        from uuid import uuid4

        from zenml.image_builders import (
            LocalImageBuilder,
            LocalImageBuilderConfig,
            LocalImageBuilderFlavor,
        )

        flavor = LocalImageBuilderFlavor()

        now = utc_now()
        image_builder = LocalImageBuilder(
            id=uuid4(),
            name="temporary_default",
            flavor=flavor.name,
            type=flavor.type,
            config=LocalImageBuilderConfig(),
            environment={},
            user=Client().active_user.id,
            created=now,
            updated=now,
            secrets=[],
        )

        self._components[StackComponentType.IMAGE_BUILDER] = [
            image_builder
        ]
validate_log_store() -> None

Validates that the stack has a log store.

Source code in src/zenml/stack/stack.py 
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
def validate_log_store(self) -> None:
    """Validates that the stack has a log store."""
    if self._get_default_component(StackComponentType.LOG_STORE):
        return

    from zenml.log_stores import ArtifactLogStore

    self._components[StackComponentType.LOG_STORE] = [
        ArtifactLogStore.from_artifact_store(self.artifact_store),
    ]
Functions
Modules

stack_component

Implementation of the ZenML Stack Component class.

Classes
StackComponent(name: str, id: UUID, config: StackComponentConfig, flavor: str, type: StackComponentType, user: Optional[UUID], created: datetime, updated: datetime, environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None, labels: Optional[Dict[str, Any]] = None, connector_requirements: Optional[ServiceConnectorRequirements] = None, connector: Optional[UUID] = None, connector_resource_id: Optional[str] = None, *args: Any, **kwargs: Any)

Abstract StackComponent class for all components of a ZenML stack.

Initializes a StackComponent.

Parameters:

Name Type Description Default
name str

The name of the component.

 required
id UUID

The unique ID of the component.

 required
config StackComponentConfig

The config of the component.

 required
flavor str

The flavor of the component.

 required
type StackComponentType

The type of the component.

 required
user Optional[UUID]

The ID of the user who created the component.

 required
created datetime

The creation time of the component.

 required
updated datetime

The last update time of the component.

 required
environment Optional[Dict[str, str]]

Environment variables to set when running on this component.

 None
secrets Optional[List[UUID]]

Secrets to set as environment variables when running on this component.

 None
labels Optional[Dict[str, Any]]

The labels of the component.

 None
connector_requirements Optional[ServiceConnectorRequirements]

The requirements for the connector.

 None
connector Optional[UUID]

The ID of a connector linked to the component.

 None
connector_resource_id Optional[str]

The custom resource ID to access through the connector.

 None
*args Any

Additional positional arguments.

 ()
**kwargs Any

Additional keyword arguments.

 {}

Raises:

Type Description
ValueError

If a secret reference is passed as name.
Source code in src/zenml/stack/stack_component.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
def __init__(
    self,
    name: str,
    id: UUID,
    config: StackComponentConfig,
    flavor: str,
    type: StackComponentType,
    user: Optional[UUID],
    created: datetime,
    updated: datetime,
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
    labels: Optional[Dict[str, Any]] = None,
    connector_requirements: Optional[ServiceConnectorRequirements] = None,
    connector: Optional[UUID] = None,
    connector_resource_id: Optional[str] = None,
    *args: Any,
    **kwargs: Any,
):
    """Initializes a StackComponent.

    Args:
        name: The name of the component.
        id: The unique ID of the component.
        config: The config of the component.
        flavor: The flavor of the component.
        type: The type of the component.
        user: The ID of the user who created the component.
        created: The creation time of the component.
        updated: The last update time of the component.
        environment: Environment variables to set when running on this
            component.
        secrets: Secrets to set as environment variables when running on
            this component.
        labels: The labels of the component.
        connector_requirements: The requirements for the connector.
        connector: The ID of a connector linked to the component.
        connector_resource_id: The custom resource ID to access through
            the connector.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Raises:
        ValueError: If a secret reference is passed as name.
    """
    if secret_utils.is_secret_reference(name):
        raise ValueError(
            "Passing the `name` attribute of a stack component as a "
            "secret reference is not allowed."
        )

    self.id = id
    self.name = name
    self._config = config
    self.flavor = flavor
    self.type = type
    self.user = user
    self.created = created
    self.updated = updated
    self.labels = labels
    self.environment = environment or {}
    self.secrets = secrets or []
    self.connector_requirements = connector_requirements
    self.connector = connector
    self.connector_resource_id = connector_resource_id
    self._connector_instance: Optional[ServiceConnector] = None
Attributes
apt_packages: List[str] property

List of APT package requirements for the component.

Returns:

Type Description
List[str]

A list of APT package requirements for the component.
config: StackComponentConfig property

Returns the configuration of the stack component.

This should be overwritten by any subclasses that define custom configs to return the correct config class.

Returns:

Type Description
StackComponentConfig

The configuration of the stack component.
local_path: Optional[str] property

Path to a local directory to store persistent information.

This property should only be implemented by components that need to store persistent information in a directory on the local machine and also need that information to be available during pipeline runs.

IMPORTANT: the path returned by this property must always be a path that is relative to the ZenML local store's directory. The local orchestrators rely on this convention to correctly mount the local folders in the containers. This is an example of a valid path:

from zenml.config.global_config import GlobalConfiguration

...

@property
def local_path(self) -> Optional[str]:

    return os.path.join(
        GlobalConfiguration().local_stores_path,
        str(self.uuid),
    )

Returns:

Type Description
Optional[str]

A path to a local directory used by the component to store
Optional[str]

persistent information.
log_file: Optional[str] property

Optional path to a log file for the stack component.

Returns:

Type Description
Optional[str]

Optional path to a log file for the stack component.
post_registration_message: Optional[str] property

Optional message printed after the stack component is registered.

Returns:

Type Description
Optional[str]

An optional message.
requirements: Set[str] property

Set of PyPI requirements for the component.

Returns:

Type Description
Set[str]

A set of PyPI requirements for the component.
settings_class: Optional[Type[BaseSettings]] property

Class specifying available settings for this component.

Returns:

Type Description
Optional[Type[BaseSettings]]

Optional settings class.
validator: Optional[StackValidator] property

The optional validator of the stack component.

This validator will be called each time a stack with the stack component is initialized. Subclasses should override this property and return a StackValidator that makes sure they're not included in any stack that they're not compatible with.

Returns:

Type Description
Optional[StackValidator]

An optional StackValidator instance.
Functions
cleanup() -> None

Cleans up the component after it has been used.

Source code in src/zenml/stack/stack_component.py 
794
795
796
def cleanup(self) -> None:
    """Cleans up the component after it has been used."""
    pass
cleanup_step_run(info: StepRunInfo, step_failed: bool) -> None

Cleans up resources after the step run is finished.

Parameters:

Name Type Description Default
info StepRunInfo

Info about the step that was executed.

 required
step_failed bool

Whether the step failed.

 required
Source code in src/zenml/stack/stack_component.py 
763
764
765
766
767
768
769
def cleanup_step_run(self, info: "StepRunInfo", step_failed: bool) -> None:
    """Cleans up resources after the step run is finished.

    Args:
        info: Info about the step that was executed.
        step_failed: Whether the step failed.
    """
connector_has_expired() -> bool

Checks whether the connector linked to this stack component has expired.

Returns:

Type Description
bool

Whether the connector linked to this stack component has expired, or isn't linked to a connector.
Source code in src/zenml/stack/stack_component.py 
562
563
564
565
566
567
568
569
570
571
572
573
574
575
def connector_has_expired(self) -> bool:
    """Checks whether the connector linked to this stack component has expired.

    Returns:
        Whether the connector linked to this stack component has expired, or isn't linked to a connector.
    """
    if self.connector is None:
        # The stack component isn't linked to a connector
        return False

    if self._connector_instance is None:
        return True

    return self._connector_instance.has_expired()
from_model(component_model: ComponentResponse) -> StackComponent classmethod

Creates a StackComponent from a ComponentModel.

Parameters:

Name Type Description Default
component_model ComponentResponse

The ComponentModel to create the StackComponent

 required

Returns:

Type Description
StackComponent

The created StackComponent.

Raises:

Type Description
ImportError

If the flavor can't be imported.
Source code in src/zenml/stack/stack_component.py 
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
@classmethod
def from_model(
    cls, component_model: "ComponentResponse"
) -> "StackComponent":
    """Creates a StackComponent from a ComponentModel.

    Args:
        component_model: The ComponentModel to create the StackComponent

    Returns:
        The created StackComponent.

    Raises:
        ImportError: If the flavor can't be imported.
    """
    from zenml.stack import Flavor

    flavor_model = component_model.flavor
    flavor = Flavor.from_model(flavor_model)

    configuration = flavor.config_class(**component_model.configuration)

    user_id = component_model.user_id

    try:
        return flavor.implementation_class(
            user=user_id,
            name=component_model.name,
            id=component_model.id,
            config=configuration,
            environment=component_model.environment,
            secrets=component_model.secrets,
            labels=component_model.labels,
            flavor=component_model.flavor_name,
            type=component_model.type,
            created=component_model.created,
            updated=component_model.updated,
            connector_requirements=flavor.service_connector_requirements,
            connector=component_model.connector.id
            if component_model.connector
            else None,
            connector_resource_id=component_model.connector_resource_id,
        )
    except ImportError as e:
        from zenml.integrations.registry import integration_registry

        integration_requirements = " ".join(
            f"'{req}'"
            for req in integration_registry.select_integration_requirements(
                flavor_model.integration
            )
        )

        if integration_registry.is_installed(flavor_model.integration):
            raise ImportError(
                f"{e}\n\n"
                f"Something went wrong while trying to import from the "
                f"`{flavor_model.integration}` integration. Please make "
                "sure that all its requirements are installed properly by "
                "reinstalling the integration either through our CLI: "
                f"`zenml integration install {flavor_model.integration} "
                "-y` or by manually installing its requirements: "
                f"`pip install {integration_requirements}`. If the error "
                "persists, please contact the ZenML team."
            ) from e
        else:
            raise ImportError(
                f"{e}\n\n"
                f"The `{flavor_model.integration}` integration that you "
                "are trying to use is not installed in your current "
                "environment. Please make sure that it is installed by "
                "either using our CLI: `zenml integration install "
                f"{flavor_model.integration}` or by manually installing "
                f"its requirements: `pip install "
                f"{integration_requirements}`"
            ) from e
get_connector() -> Optional[ServiceConnector]

Returns the connector linked to this stack component.

Returns:

Type Description
Optional[ServiceConnector]

The connector linked to this stack component.

Raises:

Type Description
RuntimeError

If the stack component does not specify connector requirements or if the connector linked to the component is not compatible or not found.
Source code in src/zenml/stack/stack_component.py 
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
def get_connector(self) -> Optional["ServiceConnector"]:
    """Returns the connector linked to this stack component.

    Returns:
        The connector linked to this stack component.

    Raises:
        RuntimeError: If the stack component does not specify connector
            requirements or if the connector linked to the component is not
            compatible or not found.
    """
    from zenml.client import Client

    if self.connector is None:
        return None

    if self._connector_instance is not None:
        # If the connector instance is still valid, return it. Otherwise,
        # we'll try to get a new one.
        if not self._connector_instance.has_expired():
            return self._connector_instance

    if self.connector_requirements is None:
        raise RuntimeError(
            f"Unable to get connector for component {self} because this "
            "component does not declare any connector requirements in its. "
            "flavor specification. Override the "
            "`service_connector_requirements` method in its flavor class "
            "to return a connector requirements specification and try "
            "again."
        )

    if self.connector_requirements.resource_id_attr is not None:
        # Check if an attribute is set in the component configuration
        resource_id = getattr(
            self.config, self.connector_requirements.resource_id_attr
        )
    else:
        # Otherwise, use the resource ID configured in the component
        resource_id = self.connector_resource_id

    client = Client()
    try:
        self._connector_instance = client.get_service_connector_client(
            name_id_or_prefix=self.connector,
            resource_type=self.connector_requirements.resource_type,
            resource_id=resource_id,
        )
    except KeyError:
        raise RuntimeError(
            f"The connector with ID {self.connector} linked "
            f"to the '{self.name}' {self.type} stack component could not "
            f"be found or is not accessible. Please verify that the "
            f"connector exists and that you have access to it."
        )
    except ValueError as e:
        raise RuntimeError(
            f"The connector with ID {self.connector} linked "
            f"to the '{self.name}' {self.type} stack component could not "
            f"be correctly configured: {e}."
        )
    except AuthorizationException as e:
        raise RuntimeError(
            f"The connector with ID {self.connector} linked "
            f"to the '{self.name}' {self.type} stack component could not "
            f"be accessed due to an authorization error: {e}. Please "
            f"verify that you have access to the connector and try again."
        )

    return self._connector_instance
get_docker_builds(snapshot: PipelineSnapshotBase) -> List[BuildConfiguration]

Gets the Docker builds required for the component.

Parameters:

Name Type Description Default
snapshot PipelineSnapshotBase

The pipeline snapshot for which to get the builds.

 required

Returns:

Type Description
List[BuildConfiguration]

The required Docker builds.
Source code in src/zenml/stack/stack_component.py 
717
718
719
720
721
722
723
724
725
726
727
728
def get_docker_builds(
    self, snapshot: "PipelineSnapshotBase"
) -> List["BuildConfiguration"]:
    """Gets the Docker builds required for the component.

    Args:
        snapshot: The pipeline snapshot for which to get the builds.

    Returns:
        The required Docker builds.
    """
    return []
get_pipeline_run_metadata(run_id: UUID) -> Dict[str, MetadataType]

Get general component-specific metadata for a pipeline run.

Parameters:

Name Type Description Default
run_id UUID

The ID of the pipeline run.

 required

Returns:

Type Description
Dict[str, MetadataType]

A dictionary of metadata.
Source code in src/zenml/stack/stack_component.py 
730
731
732
733
734
735
736
737
738
739
740
741
def get_pipeline_run_metadata(
    self, run_id: UUID
) -> Dict[str, "MetadataType"]:
    """Get general component-specific metadata for a pipeline run.

    Args:
        run_id: The ID of the pipeline run.

    Returns:
        A dictionary of metadata.
    """
    return {}
get_settings(container: Union[Step, StepRunResponse, StepRunInfo, PipelineSnapshotBase, PipelineSnapshotResponse, PipelineRunResponse]) -> BaseSettings

Gets settings for this stack component.

This will return None if the stack component doesn't specify a settings class or the container doesn't contain runtime options for this component.

Parameters:

Name Type Description Default
container Union[Step, StepRunResponse, StepRunInfo, PipelineSnapshotBase, PipelineSnapshotResponse, PipelineRunResponse]

The Step, StepRunInfo or PipelineSnapshot from which to get the settings.

 required

Returns:

Type Description
BaseSettings

Settings for this stack component.

Raises:

Type Description
RuntimeError

If the stack component does not specify a settings class.
Source code in src/zenml/stack/stack_component.py 
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
def get_settings(
    self,
    container: Union[
        "Step",
        "StepRunResponse",
        "StepRunInfo",
        "PipelineSnapshotBase",
        "PipelineSnapshotResponse",
        "PipelineRunResponse",
    ],
) -> "BaseSettings":
    """Gets settings for this stack component.

    This will return `None` if the stack component doesn't specify a
    settings class or the container doesn't contain runtime
    options for this component.

    Args:
        container: The `Step`, `StepRunInfo` or `PipelineSnapshot` from
            which to get the settings.

    Returns:
        Settings for this stack component.

    Raises:
        RuntimeError: If the stack component does not specify a settings
            class.
    """
    if not self.settings_class:
        raise RuntimeError(
            f"Unable to get settings for component {self} because this "
            "component does not have an associated settings class. "
            "Return a settings class from the `@settings_class` property "
            "and try again."
        )

    canonical_key = settings_utils.get_stack_component_name_setting_key(
        self
    )
    legacy_key = settings_utils.get_stack_component_setting_key(self)

    all_settings = (
        container.config.settings
        if isinstance(
            container,
            (Step, StepRunResponse, StepRunInfo, PipelineRunResponse),
        )
        else container.pipeline_configuration.settings
    )

    # Use the current config as a base
    settings_dict = self.config.model_dump(exclude_unset=True)

    if canonical_key in all_settings:
        settings_dict.update(dict(all_settings[canonical_key]))
    elif legacy_key in all_settings:
        settings_dict.update(dict(all_settings[legacy_key]))

    return self.settings_class.model_validate(settings_dict)
get_step_run_metadata(info: StepRunInfo) -> Dict[str, MetadataType]

Get component- and step-specific metadata after a step ran.

Parameters:

Name Type Description Default
info StepRunInfo

Info about the step that was executed.

 required

Returns:

Type Description
Dict[str, MetadataType]

A dictionary of metadata.
Source code in src/zenml/stack/stack_component.py 
750
751
752
753
754
755
756
757
758
759
760
761
def get_step_run_metadata(
    self, info: "StepRunInfo"
) -> Dict[str, "MetadataType"]:
    """Get component- and step-specific metadata after a step ran.

    Args:
        info: Info about the step that was executed.

    Returns:
        A dictionary of metadata.
    """
    return {}
prepare_step_run(info: StepRunInfo) -> None

Prepares running a step.

Parameters:

Name Type Description Default
info StepRunInfo

Info about the step that will be executed.

 required
Source code in src/zenml/stack/stack_component.py 
743
744
745
746
747
748
def prepare_step_run(self, info: "StepRunInfo") -> None:
    """Prepares running a step.

    Args:
        info: Info about the step that will be executed.
    """
StackComponentConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: BaseModel, ABC

Base class for all ZenML stack component configs.

Ensures that secret references don't clash with pydantic validation.

StackComponents allow the specification of all their string attributes using secret references of the form {{secret_name.key}}. This however is only possible when the stack component does not perform any explicit validation of this attribute using pydantic validators. If this were the case, the validation would run on the secret reference and would fail or in the worst case, modify the secret reference and lead to unexpected behavior. This method ensures that no attributes that require custom pydantic validation are set as secret references.

Parameters:

Name Type Description Default
warn_about_plain_text_secrets bool

If true, then warns about using plain-text secrets.

 False
**kwargs Any

Arguments to initialize this stack component.

 {}

Raises:

Type Description
ValueError

If an attribute that requires custom pydantic validation is passed as a secret reference, or if the name attribute was passed as a secret reference.
Source code in src/zenml/stack/stack_component.py 
 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references don't clash with pydantic validation.

    StackComponents allow the specification of all their string attributes
    using secret references of the form `{{secret_name.key}}`. This however
    is only possible when the stack component does not perform any explicit
    validation of this attribute using pydantic validators. If this were
    the case, the validation would run on the secret reference and would
    fail or in the worst case, modify the secret reference and lead to
    unexpected behavior. This method ensures that no attributes that require
    custom pydantic validation are set as secret references.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using
            plain-text secrets.
        **kwargs: Arguments to initialize this stack component.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            is passed as a secret reference, or if the `name` attribute
            was passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}` for a `{self.__class__.__name__}` "
                    "stack component. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure your stack "
                    "components with secrets here: "
                    "https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if pydantic_utils.has_validators(
            pydantic_class=self.__class__, field_name=key
        ):
            raise ValueError(
                f"Passing the stack component attribute `{key}` as a "
                "secret reference is not allowed as additional validation "
                "is required for this attribute."
            )

    super().__init__(**kwargs)
Attributes
is_local: bool property

Checks if this stack component is running locally.

Concrete stack component configuration classes should override this method to return True if the stack component is relying on local resources or capabilities (e.g. local filesystem, local database or other services).

Examples:

  • Artifact Stores that store artifacts in the local filesystem
  • Orchestrators that are connected to local orchestration runtime services (e.g. local Kubernetes clusters, Docker containers etc).

Returns:

Type Description
bool

True if this config is for a local component, False otherwise.
is_remote: bool property

Checks if this stack component is running remotely.

Concrete stack component configuration classes should override this method to return True if the stack component is running in a remote location, and it needs to access the ZenML database.

This designation is used to determine if the stack component can be used with a local ZenML database or if it requires a remote ZenML server.

Examples:

  • Orchestrators that are running pipelines in the cloud or in a location other than the local host
  • Step Operators that are running steps in the cloud or in a location other than the local host

Returns:

Type Description
bool

True if this config is for a remote component, False otherwise.
is_valid: bool property

Checks if the stack component configurations are valid.

Concrete stack component configuration classes should override this method to return False if the stack component configurations are invalid.

Returns:

Type Description
bool

True if the stack component config is valid, False otherwise.
required_secrets: Set[secret_utils.SecretReference] property

All required secrets for this stack component.

Returns:

Type Description
Set[SecretReference]

The required secrets of this stack component.
Functions
Functions
Modules

stack_validator

Implementation of the ZenML Stack Validator.

Classes
StackValidator(required_components: Optional[AbstractSet[StackComponentType]] = None, custom_validation_function: Optional[Callable[[Stack], Tuple[bool, str]]] = None)

A StackValidator is used to validate a stack configuration.

Each StackComponent can provide a StackValidator to make sure it is compatible with all components of the stack. The KubeflowOrchestrator for example will always require the stack to have a container registry in order to push the docker images that are required to run a pipeline in Kubeflow Pipelines.

Initializes a StackValidator instance.

Parameters:

Name Type Description Default
required_components Optional[AbstractSet[StackComponentType]]

Optional set of stack components that must exist in the stack.

 None
custom_validation_function Optional[Callable[[Stack], Tuple[bool, str]]]

Optional function that returns whether a stack is valid and an error message to show if not valid.

 None
Source code in src/zenml/stack/stack_validator.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(
    self,
    required_components: Optional[AbstractSet[StackComponentType]] = None,
    custom_validation_function: Optional[
        Callable[["Stack"], Tuple[bool, str]]
    ] = None,
):
    """Initializes a `StackValidator` instance.

    Args:
        required_components: Optional set of stack components that must
            exist in the stack.
        custom_validation_function: Optional function that returns whether
            a stack is valid and an error message to show if not valid.
    """
    self._required_components = required_components or set()
    self._custom_validation_function = custom_validation_function
Functions
validate(stack: Stack) -> None

Validates the given stack.

Checks if the stack contains all the required components and passes the custom validation function of the validator.

Parameters:

Name Type Description Default
stack Stack

The stack to validate.

 required

Raises:

Type Description
StackValidationError

If the stack does not meet all the validation criteria.
Source code in src/zenml/stack/stack_validator.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def validate(self, stack: "Stack") -> None:
    """Validates the given stack.

    Checks if the stack contains all the required components and passes
    the custom validation function of the validator.

    Args:
        stack: The stack to validate.

    Raises:
        StackValidationError: If the stack does not meet all the
            validation criteria.
    """
    existing_components = set(c.type for c in stack.all_components)

    missing_components = self._required_components - existing_components
    if missing_components:
        raise StackValidationError(
            f"Missing stack components {missing_components} for "
            f"stack: {stack.name}"
        )

    if self._custom_validation_function:
        valid, err_msg = self._custom_validation_function(stack)
        if not valid:
            raise StackValidationError(
                f"Custom validation function failed to validate "
                f"stack '{stack.name}': {err_msg}"
            )
Functions

utils

Util functions for handling stacks, components, and flavors.

Classes
Functions
get_flavor_by_name_and_type_from_zen_store(zen_store: BaseZenStore, flavor_name: str, component_type: StackComponentType) -> FlavorResponse

Get a stack component flavor by name and type from a ZenStore.

Parameters:

Name Type Description Default
zen_store BaseZenStore

The ZenStore to query.

 required
flavor_name str

The name of a stack component flavor.

 required
component_type StackComponentType

The type of the stack component.

 required

Returns:

Type Description
FlavorResponse

The flavor model.

Raises:

Type Description
KeyError

If no flavor with the given name and type exists.
Source code in src/zenml/stack/utils.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def get_flavor_by_name_and_type_from_zen_store(
    zen_store: BaseZenStore,
    flavor_name: str,
    component_type: StackComponentType,
) -> FlavorResponse:
    """Get a stack component flavor by name and type from a ZenStore.

    Args:
        zen_store: The ZenStore to query.
        flavor_name: The name of a stack component flavor.
        component_type: The type of the stack component.

    Returns:
        The flavor model.

    Raises:
        KeyError: If no flavor with the given name and type exists.
    """
    flavors = zen_store.list_flavors(
        FlavorFilter(name=flavor_name, type=component_type)
    )
    if not flavors:
        raise KeyError(
            f"No flavor with name '{flavor_name}' and type "
            f"'{component_type}' exists."
        )
    return flavors[0]
validate_stack_component_config(configuration_dict: Dict[str, Any], flavor: Union[FlavorResponse, str], component_type: StackComponentType, zen_store: Optional[BaseZenStore] = None, validate_custom_flavors: bool = True, existing_config: Optional[Dict[str, Any]] = None) -> Optional[StackComponentConfig]

Validate the configuration of a stack component.

Parameters:

Name Type Description Default
configuration_dict Dict[str, Any]

The stack component configuration to validate.

 required
flavor Union[FlavorResponse, str]

The model or name of the flavor of the stack component.

 required
component_type StackComponentType

The type of the stack component.

 required
zen_store Optional[BaseZenStore]

An optional ZenStore in which to look for the flavor. If not provided, the flavor will be fetched via the regular ZenML Client. This is mainly useful for checks running inside the ZenML server.

 None
validate_custom_flavors bool

When loading custom flavors from the local environment, this flag decides whether the import failures are raised or an empty value is returned.

 True
existing_config Optional[Dict[str, Any]]

The existing stack component configuration.

 None

Returns:

Type Description
Optional[StackComponentConfig]

The validated stack component configuration or None, if the
Optional[StackComponentConfig]

flavor is a custom flavor that could not be imported from the local
Optional[StackComponentConfig]

environment and the validate_custom_flavors flag is set to False.

Raises:

Type Description
ValueError

If the configuration is invalid.
ImportError

If the flavor class could not be imported.
ModuleNotFoundError

If the flavor class could not be imported.
Source code in src/zenml/stack/utils.py 
 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
 66
 67
 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
def validate_stack_component_config(
    configuration_dict: Dict[str, Any],
    flavor: Union[FlavorResponse, str],
    component_type: StackComponentType,
    zen_store: Optional[BaseZenStore] = None,
    validate_custom_flavors: bool = True,
    existing_config: Optional[Dict[str, Any]] = None,
) -> Optional[StackComponentConfig]:
    """Validate the configuration of a stack component.

    Args:
        configuration_dict: The stack component configuration to validate.
        flavor: The model or name of the flavor of the stack component.
        component_type: The type of the stack component.
        zen_store: An optional ZenStore in which to look for the flavor. If not
            provided, the flavor will be fetched via the regular ZenML Client.
            This is mainly useful for checks running inside the ZenML server.
        validate_custom_flavors: When loading custom flavors from the local
            environment, this flag decides whether the import failures are
            raised or an empty value is returned.
        existing_config: The existing stack component configuration.

    Returns:
        The validated stack component configuration or None, if the
        flavor is a custom flavor that could not be imported from the local
        environment and the `validate_custom_flavors` flag is set to False.

    Raises:
        ValueError: If the configuration is invalid.
        ImportError: If the flavor class could not be imported.
        ModuleNotFoundError: If the flavor class could not be imported.
    """
    if isinstance(flavor, FlavorResponse):
        flavor_model = flavor
    else:
        if zen_store:
            flavor_model = get_flavor_by_name_and_type_from_zen_store(
                zen_store=zen_store,
                flavor_name=flavor,
                component_type=component_type,
            )
        else:
            flavor_model = Client().get_flavor_by_name_and_type(
                name=flavor,
                component_type=component_type,
            )

    try:
        flavor_class = Flavor.from_model(flavor_model)
    except (ImportError, ModuleNotFoundError):
        # The flavor class couldn't be loaded.
        if flavor_model.is_custom and not validate_custom_flavors:
            return None
        raise

    config_class = flavor_class.config_class

    extra_keys = set(configuration_dict) - set(config_class.model_fields)
    if existing_config:
        # If there is an existing configuration, don't fail because of keys
        # that were already set. This can be the case if the config stored in
        # the DB contains attributes that have been removed.
        extra_keys = extra_keys - set(existing_config)

    if extra_keys:
        raise ValueError(
            f"Invalid configuration keys {list(extra_keys)} for "
            f"{flavor_model.name} {component_type}."
        )

    configuration = config_class(**configuration_dict)

    if not configuration.is_valid:
        raise ValueError(
            f"Invalid stack component configuration. Please verify "
            f"the configurations set for {component_type}."
        )
    return configuration
warn_if_config_server_mismatch(configuration: StackComponentConfig) -> None

Warn if a component configuration is mismatched with the ZenML server.

Parameters:

Name Type Description Default
configuration StackComponentConfig

The component configuration to check.

 required
Source code in src/zenml/stack/utils.py 
109
110
111
112
113
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
def warn_if_config_server_mismatch(
    configuration: StackComponentConfig,
) -> None:
    """Warn if a component configuration is mismatched with the ZenML server.

    Args:
        configuration: The component configuration to check.
    """
    zen_store = Client().zen_store
    if configuration.is_remote and zen_store.is_local_store():
        if zen_store.type == StoreType.REST:
            logger.warning(
                "You are configuring a stack component that is running "
                "remotely while using a local ZenML server. The component "
                "may not be able to reach the local ZenML server and will "
                "therefore not be functional. Please consider deploying "
                "and/or using a remote ZenML server instead."
            )
        else:
            logger.warning(
                "You are configuring a stack component that is running "
                "remotely while connected to the local database. The component "
                "will not be able to reach the local database and will "
                "therefore not be functional. Please consider deploying "
                "and/or using a remote ZenML server instead."
            )
    elif configuration.is_local and not zen_store.is_local_store():
        logger.warning(
            "You are configuring a stack component that is using "
            "local resources while connected to a remote ZenML server. The "
            "stack component may not be usable from other hosts or by "
            "other users. You should consider using a non-local stack "
            "component alternative instead."
        )