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)

Flavor

Class for ZenML Flavors.

Source code in src/zenml/stack/flavor.py

class Flavor:
    """Class for ZenML Flavors."""

    @property
    @abstractmethod
    def name(self) -> str:
        """The flavor name.

        Returns:
            The flavor name.
        """

    @property
    def docs_url(self) -> Optional[str]:
        """A url to point at docs explaining this flavor.

        Returns:
            A flavor docs url.
        """
        return None

    @property
    def sdk_docs_url(self) -> Optional[str]:
        """A url to point at SDK docs explaining this flavor.

        Returns:
            A flavor SDK docs url.
        """
        return None

    @property
    def logo_url(self) -> Optional[str]:
        """A url to represent the flavor in the dashboard.

        Returns:
            The flavor logo.
        """
        return None

    @property
    @abstractmethod
    def type(self) -> StackComponentType:
        """The stack component type.

        Returns:
            The stack component type.
        """

    @property
    @abstractmethod
    def implementation_class(self) -> Type[StackComponent]:
        """Implementation class for this flavor.

        Returns:
            The implementation class for this flavor.
        """

    @property
    @abstractmethod
    def config_class(self) -> Type[StackComponentConfig]:
        """Returns `StackComponentConfig` config class.

        Returns:
            The config class.
        """

    @property
    def config_schema(self) -> Dict[str, Any]:
        """The config schema for a flavor.

        Returns:
            The config schema.
        """
        return self.config_class.model_json_schema()

    @property
    def service_connector_requirements(
        self,
    ) -> Optional[ServiceConnectorRequirements]:
        """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:
            Requirements for compatible service connectors, if a service
            connector is required for this flavor.
        """
        return None

    @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)

    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. Custom flavors
                are then scoped by user and workspace

        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
        )
        user = None
        workspace = None
        if is_custom:
            user = Client().active_user.id
            workspace = Client().active_workspace.id

        model_class = FlavorRequest if is_custom else InternalFlavorRequest
        model = model_class(
            user=user,
            workspace=workspace,
            name=self.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

    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
        """
        from zenml import __version__

        component_type = self.type.plural.replace("_", "-")
        name = self.name.replace("_", "-")

        try:
            is_latest = is_latest_zenml_version()
        except RuntimeError:
            # We assume in error cases that we are on the latest version
            is_latest = True

        if is_latest:
            base = "https://docs.zenml.io"
        else:
            base = f"https://zenml-io.gitbook.io/zenml-legacy-documentation/v/{__version__}"
        return f"{base}/stack-components/{component_type}/{name}"

    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]

            return (
                f"{base}/integration_code_docs"
                f"/integrations-{integration}/#{self.__module__}"
            )

        else:
            return (
                f"{base}/core_code_docs/core-{component_type}/"
                f"#{self.__module__}"
            )

config_class abstractmethod property

Returns StackComponentConfig config class.

Returns:

Type	Description
Type[StackComponentConfig]	The config class.

config_schema property

The config schema for a flavor.

Returns:

Type	Description
Dict[str, Any]	The config schema.

docs_url property

A url to point at docs explaining this flavor.

Returns:

Type	Description
Optional[str]	A flavor docs url.

implementation_class abstractmethod property

Implementation class for this flavor.

Returns:

Type	Description
Type[StackComponent]	The implementation class for this flavor.

logo_url property

A url to represent the flavor in the dashboard.

Returns:

Type	Description
Optional[str]	The flavor logo.

name abstractmethod property

The flavor name.

Returns:

Type	Description
str	The flavor name.

sdk_docs_url property

A url to point at SDK docs explaining this flavor.

Returns:

Type	Description
Optional[str]	A flavor SDK docs url.

service_connector_requirements 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 abstractmethod property

The stack component type.

Returns:

Type	Description
StackComponentType	The stack component type.

from_model(flavor_model) 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

@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()

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

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
    """
    from zenml import __version__

    component_type = self.type.plural.replace("_", "-")
    name = self.name.replace("_", "-")

    try:
        is_latest = is_latest_zenml_version()
    except RuntimeError:
        # We assume in error cases that we are on the latest version
        is_latest = True

    if is_latest:
        base = "https://docs.zenml.io"
    else:
        base = f"https://zenml-io.gitbook.io/zenml-legacy-documentation/v/{__version__}"
    return f"{base}/stack-components/{component_type}/{name}"

generate_default_sdk_docs_url()

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

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]

        return (
            f"{base}/integration_code_docs"
            f"/integrations-{integration}/#{self.__module__}"
        )

    else:
        return (
            f"{base}/core_code_docs/core-{component_type}/"
            f"#{self.__module__}"
        )

to_model(integration=None, is_custom=True)

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. Custom flavors are then scoped by user and workspace	True

Returns:

Type	Description
FlavorRequest	The model.

Source code in src/zenml/stack/flavor.py

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. Custom flavors
            are then scoped by user and workspace

    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
    )
    user = None
    workspace = None
    if is_custom:
        user = Client().active_user.id
        workspace = Client().active_workspace.id

    model_class = FlavorRequest if is_custom else InternalFlavorRequest
    model = model_class(
        user=user,
        workspace=workspace,
        name=self.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

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.

Source code in src/zenml/stack/stack.py

class Stack:
    """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.
    """

    def __init__(
        self,
        id: UUID,
        name: str,
        *,
        orchestrator: "BaseOrchestrator",
        artifact_store: "BaseArtifactStore",
        container_registry: Optional["BaseContainerRegistry"] = None,
        step_operator: Optional["BaseStepOperator"] = None,
        feature_store: Optional["BaseFeatureStore"] = None,
        model_deployer: Optional["BaseModelDeployer"] = None,
        experiment_tracker: Optional["BaseExperimentTracker"] = None,
        alerter: Optional["BaseAlerter"] = None,
        annotator: Optional["BaseAnnotator"] = None,
        data_validator: Optional["BaseDataValidator"] = None,
        image_builder: Optional["BaseImageBuilder"] = None,
        model_registry: Optional["BaseModelRegistry"] = None,
    ):
        """Initializes and validates a stack instance.

        Args:
            id: Unique ID of the stack.
            name: Name of the 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.
        """
        self._id = id
        self._name = name
        self._orchestrator = orchestrator
        self._artifact_store = artifact_store
        self._container_registry = container_registry
        self._step_operator = step_operator
        self._feature_store = feature_store
        self._model_deployer = model_deployer
        self._experiment_tracker = experiment_tracker
        self._alerter = alerter
        self._annotator = annotator
        self._data_validator = data_validator
        self._model_registry = model_registry
        self._image_builder = image_builder

    @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.
        """
        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
        component_models = pagination_utils.depaginate(
            Client().list_stack_components,
            stack_id=stack_model.id,
            hydrate=True,
        )

        stack_components = {
            model.type: StackComponent.from_model(model)
            for model in component_models
        }
        stack = Stack.from_components(
            id=stack_model.id,
            name=stack_model.name,
            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

    @classmethod
    def from_components(
        cls,
        id: UUID,
        name: str,
        components: Dict[StackComponentType, "StackComponent"],
    ) -> "Stack":
        """Creates a stack instance from a dict of stack components.

        # noqa: DAR402

        Args:
            id: Unique ID of the stack.
            name: The name of the stack.
            components: The components of the 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.
        """
        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.experiment_trackers import BaseExperimentTracker
        from zenml.feature_stores import BaseFeatureStore
        from zenml.image_builders import BaseImageBuilder
        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)

        return Stack(
            id=id,
            name=name,
            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,
        )

    @property
    def components(self) -> Dict[StackComponentType, "StackComponent"]:
        """All components of the stack.

        Returns:
            A dictionary of all components of the stack.
        """
        return {
            component.type: component
            for component in [
                self.orchestrator,
                self.artifact_store,
                self.container_registry,
                self.step_operator,
                self.feature_store,
                self.model_deployer,
                self.experiment_tracker,
                self.alerter,
                self.annotator,
                self.data_validator,
                self.image_builder,
                self.model_registry,
            ]
            if component is not None
        }

    @property
    def id(self) -> UUID:
        """The ID of the stack.

        Returns:
            The ID of the stack.
        """
        return self._id

    @property
    def name(self) -> str:
        """The name of the stack.

        Returns:
            str: The name of the stack.
        """
        return self._name

    @property
    def orchestrator(self) -> "BaseOrchestrator":
        """The orchestrator of the stack.

        Returns:
            The orchestrator of the stack.
        """
        return self._orchestrator

    @property
    def artifact_store(self) -> "BaseArtifactStore":
        """The artifact store of the stack.

        Returns:
            The artifact store of the stack.
        """
        return self._artifact_store

    @property
    def container_registry(self) -> Optional["BaseContainerRegistry"]:
        """The container registry of the stack.

        Returns:
            The container registry of the stack or None if the stack does not
            have a container registry.
        """
        return self._container_registry

    @property
    def step_operator(self) -> Optional["BaseStepOperator"]:
        """The step operator of the stack.

        Returns:
            The step operator of the stack.
        """
        return self._step_operator

    @property
    def feature_store(self) -> Optional["BaseFeatureStore"]:
        """The feature store of the stack.

        Returns:
            The feature store of the stack.
        """
        return self._feature_store

    @property
    def model_deployer(self) -> Optional["BaseModelDeployer"]:
        """The model deployer of the stack.

        Returns:
            The model deployer of the stack.
        """
        return self._model_deployer

    @property
    def experiment_tracker(self) -> Optional["BaseExperimentTracker"]:
        """The experiment tracker of the stack.

        Returns:
            The experiment tracker of the stack.
        """
        return self._experiment_tracker

    @property
    def alerter(self) -> Optional["BaseAlerter"]:
        """The alerter of the stack.

        Returns:
            The alerter of the stack.
        """
        return self._alerter

    @property
    def annotator(self) -> Optional["BaseAnnotator"]:
        """The annotator of the stack.

        Returns:
            The annotator of the stack.
        """
        return self._annotator

    @property
    def data_validator(self) -> Optional["BaseDataValidator"]:
        """The data validator of the stack.

        Returns:
            The data validator of the stack.
        """
        return self._data_validator

    @property
    def image_builder(self) -> Optional["BaseImageBuilder"]:
        """The image builder of the stack.

        Returns:
            The image builder of the stack.
        """
        return self._image_builder

    @property
    def model_registry(self) -> Optional["BaseModelRegistry"]:
        """The model registry of the stack.

        Returns:
            The model registry of the stack.
        """
        return self._model_registry

    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"), sort_keys=True
            )
            for component_type, component in self.components.items()
        }
        component_dict.update({"name": self.name})
        return component_dict

    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.components.values()
            if component.type not in exclude_components
        ]
        return set.union(*requirements) if requirements else set()

    @property
    def apt_packages(self) -> List[str]:
        """List of APT package requirements for the stack.

        Returns:
            A list of APT package requirements for the stack.
        """
        return [
            package
            for component in self.components.values()
            for package in component.apt_packages
        ]

    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.components.values():
            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

    @property
    def required_secrets(self) -> Set["secret_utils.SecretReference"]:
        """All required secrets for this stack.

        Returns:
            The required secrets of this stack.
        """
        secrets = [
            component.config.required_secrets
            for component in self.components.values()
        ]
        return set.union(*secrets) if secrets else set()

    @property
    def setting_classes(self) -> Dict[str, Type["BaseSettings"]]:
        """Setting classes of all components of this stack.

        Returns:
            All setting classes and their respective keys.
        """
        setting_classes = {}
        for component in self.components.values():
            if component.settings_class:
                key = settings_utils.get_stack_component_setting_key(component)
                setting_classes[key] = component.settings_class
        return setting_classes

    @property
    def requires_remote_server(self) -> bool:
        """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:
            If the stack requires a remote ZenServer to run.
        """
        return self.orchestrator.config.is_remote or (
            self.step_operator is not None
            and self.step_operator.config.is_remote
        )

    def _validate_secrets(self, raise_exception: bool) -> None:
        """Validates that all secrets of the stack exists.

        Args:
            raise_exception: If `True`, raises an exception if a secret is
                missing. Otherwise a warning is logged.

        # noqa: DAR402
        Raises:
            StackValidationError: If a secret is missing.
        """
        env_value = os.getenv(
            ENV_ZENML_SECRET_VALIDATION_LEVEL,
            default=SecretValidationLevel.SECRET_AND_KEY_EXISTS.value,
        )
        secret_validation_level = SecretValidationLevel(env_value)

        required_secrets = self.required_secrets
        if (
            secret_validation_level != SecretValidationLevel.NONE
            and required_secrets
        ):

            def _handle_error(message: str) -> None:
                """Handles the error by raising an exception or logging.

                Args:
                    message: The error message.

                Raises:
                    StackValidationError: If called and `raise_exception` of
                        the outer method is `True`.
                """
                if raise_exception:
                    raise StackValidationError(message)
                else:
                    message += (
                        "\nYou need to solve this issue before running "
                        "a pipeline on this stack."
                    )
                    logger.warning(message)

            client = Client()

            # Attempt to resolve secrets through the secrets store
            for secret_ref in required_secrets.copy():
                try:
                    secret = client.get_secret(secret_ref.name)
                    if (
                        secret_validation_level
                        == SecretValidationLevel.SECRET_AND_KEY_EXISTS
                    ):
                        _ = secret.values[secret_ref.key]
                except (KeyError, NotImplementedError):
                    pass
                else:
                    # Drop this secret from the list of required secrets
                    required_secrets.remove(secret_ref)

            if not required_secrets:
                return

            secrets_msg = ", ".join(
                [
                    f"{secret_ref.name}.{secret_ref.key}"
                    for secret_ref in required_secrets
                ]
            )

            _handle_error(
                f"Some components in the `{self.name}` stack reference secrets "
                f"or secret keys that do not exist in the secret store: "
                f"{secrets_msg}.\nTo register the "
                "missing secrets for this stack, run `zenml stack "
                f"register-secrets {self.name}`\nIf you want to "
                "adjust the degree to which ZenML validates the existence "
                "of secrets in your stack, you can do so by setting the "
                f"environment variable {ENV_ZENML_SECRET_VALIDATION_LEVEL} "
                "to one of the following values: "
                f"{SecretValidationLevel.values()}."
            )

    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()
        for component in self.components.values():
            if component.validator:
                component.validator.validate(stack=self)

        self._validate_secrets(raise_exception=fail_if_secrets_missing)

    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.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(),
                user=Client().active_user.id,
                workspace=Client().active_workspace.id,
                created=now,
                updated=now,
            )

            self._image_builder = image_builder

    def prepare_pipeline_deployment(
        self, deployment: "PipelineDeploymentResponse"
    ) -> None:
        """Prepares the stack for a pipeline deployment.

        This method is called before a pipeline is deployed.

        Args:
            deployment: The pipeline deployment

        Raises:
            RuntimeError: If trying to deploy 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."
            )

        for component in self.components.values():
            component.prepare_pipeline_deployment(
                deployment=deployment, stack=self
            )

    def get_docker_builds(
        self, deployment: "PipelineDeploymentBase"
    ) -> List["BuildConfiguration"]:
        """Gets the Docker builds required for the stack.

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

        Returns:
            The required Docker builds.
        """
        return list(
            itertools.chain.from_iterable(
                component.get_docker_builds(deployment=deployment)
                for component in self.components.values()
            )
        )

    def deploy_pipeline(
        self,
        deployment: "PipelineDeploymentResponse",
        placeholder_run: Optional["PipelineRunResponse"] = None,
    ) -> None:
        """Deploys a pipeline on this stack.

        Args:
            deployment: The pipeline deployment.
            placeholder_run: An optional placeholder run for the deployment.
        """
        self.orchestrator.run(
            deployment=deployment, stack=self, placeholder_run=placeholder_run
        )

    def _get_active_components_for_step(
        self, step_config: "StepConfiguration"
    ) -> Dict[StackComponentType, "StackComponent"]:
        """Gets all the active stack components for a stack.

        Args:
            step_config: Configuration of the step for which to get the active
                components.

        Returns:
            Dictionary of active stack components.
        """

        def _is_active(component: "StackComponent") -> bool:
            """Checks whether a stack component is actively used in the step.

            Args:
                component: The component to check.

            Returns:
                If the component is used in this step.
            """
            if component.type == StackComponentType.STEP_OPERATOR:
                return component.name == step_config.step_operator

            if component.type == StackComponentType.EXPERIMENT_TRACKER:
                return component.name == step_config.experiment_tracker

            return True

        return {
            component_type: component
            for component_type, component in self.components.items()
            if _is_active(component)
        }

    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
        ).values():
            component.prepare_step_run(info=info)

    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.components.values():
            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

    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
        ).values():
            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

    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
        ).values():
            component.cleanup_step_run(info=info, step_failed=step_failed)

alerter property

The alerter of the stack.

Returns:

Type	Description
Optional[BaseAlerter]	The alerter of the stack.

annotator property

The annotator of the stack.

Returns:

Type	Description
Optional[BaseAnnotator]	The annotator of the stack.

apt_packages 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 property

The artifact store of the stack.

Returns:

Type	Description
BaseArtifactStore	The artifact store of the stack.

components property

All components of the stack.

Returns:

Type	Description
Dict[StackComponentType, StackComponent]	A dictionary of all components of the stack.

container_registry 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 property

The data validator of the stack.

Returns:

Type	Description
Optional[BaseDataValidator]	The data validator of the stack.

experiment_tracker property

The experiment tracker of the stack.

Returns:

Type	Description
Optional[BaseExperimentTracker]	The experiment tracker of the stack.

feature_store property

The feature store of the stack.

Returns:

Type	Description
Optional[BaseFeatureStore]	The feature store of the stack.

id property

The ID of the stack.

Returns:

Type	Description
UUID	The ID of the stack.

image_builder property

The image builder of the stack.

Returns:

Type	Description
Optional[BaseImageBuilder]	The image builder of the stack.

model_deployer property

The model deployer of the stack.

Returns:

Type	Description
Optional[BaseModelDeployer]	The model deployer of the stack.

model_registry property

The model registry of the stack.

Returns:

Type	Description
Optional[BaseModelRegistry]	The model registry of the stack.

name property

The name of the stack.

Returns:

Name	Type	Description
str	str	The name of the stack.

orchestrator property

The orchestrator of the stack.

Returns:

Type	Description
BaseOrchestrator	The orchestrator of the stack.

required_secrets property

All required secrets for this stack.

Returns:

Type	Description
Set[SecretReference]	The required secrets of this stack.

requires_remote_server 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.

setting_classes 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 property

The step operator of the stack.

Returns:

Type	Description
Optional[BaseStepOperator]	The step operator of the stack.

__init__(id, name, *, orchestrator, artifact_store, container_registry=None, step_operator=None, feature_store=None, model_deployer=None, experiment_tracker=None, alerter=None, annotator=None, data_validator=None, image_builder=None, model_registry=None)

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
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[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[BaseExperimentTracker]	Experiment tracker component of the stack.	None
alerter	Optional[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

Source code in src/zenml/stack/stack.py

def __init__(
    self,
    id: UUID,
    name: str,
    *,
    orchestrator: "BaseOrchestrator",
    artifact_store: "BaseArtifactStore",
    container_registry: Optional["BaseContainerRegistry"] = None,
    step_operator: Optional["BaseStepOperator"] = None,
    feature_store: Optional["BaseFeatureStore"] = None,
    model_deployer: Optional["BaseModelDeployer"] = None,
    experiment_tracker: Optional["BaseExperimentTracker"] = None,
    alerter: Optional["BaseAlerter"] = None,
    annotator: Optional["BaseAnnotator"] = None,
    data_validator: Optional["BaseDataValidator"] = None,
    image_builder: Optional["BaseImageBuilder"] = None,
    model_registry: Optional["BaseModelRegistry"] = None,
):
    """Initializes and validates a stack instance.

    Args:
        id: Unique ID of the stack.
        name: Name of the 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.
    """
    self._id = id
    self._name = name
    self._orchestrator = orchestrator
    self._artifact_store = artifact_store
    self._container_registry = container_registry
    self._step_operator = step_operator
    self._feature_store = feature_store
    self._model_deployer = model_deployer
    self._experiment_tracker = experiment_tracker
    self._alerter = alerter
    self._annotator = annotator
    self._data_validator = data_validator
    self._model_registry = model_registry
    self._image_builder = image_builder

check_local_paths()

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

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.components.values():
        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, step_failed)

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

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
    ).values():
        component.cleanup_step_run(info=info, step_failed=step_failed)

deploy_pipeline(deployment, placeholder_run=None)

Deploys a pipeline on this stack.

Parameters:

Name	Type	Description	Default
deployment	PipelineDeploymentResponse	The pipeline deployment.	required
placeholder_run	Optional[PipelineRunResponse]	An optional placeholder run for the deployment.	None

Source code in src/zenml/stack/stack.py

def deploy_pipeline(
    self,
    deployment: "PipelineDeploymentResponse",
    placeholder_run: Optional["PipelineRunResponse"] = None,
) -> None:
    """Deploys a pipeline on this stack.

    Args:
        deployment: The pipeline deployment.
        placeholder_run: An optional placeholder run for the deployment.
    """
    self.orchestrator.run(
        deployment=deployment, stack=self, placeholder_run=placeholder_run
    )

dict()

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

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"), sort_keys=True
        )
        for component_type, component in self.components.items()
    }
    component_dict.update({"name": self.name})
    return component_dict

from_components(id, name, components) classmethod

Creates a stack instance from a dict of stack components.

noqa: DAR402

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

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

@classmethod
def from_components(
    cls,
    id: UUID,
    name: str,
    components: Dict[StackComponentType, "StackComponent"],
) -> "Stack":
    """Creates a stack instance from a dict of stack components.

    # noqa: DAR402

    Args:
        id: Unique ID of the stack.
        name: The name of the stack.
        components: The components of the 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.
    """
    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.experiment_trackers import BaseExperimentTracker
    from zenml.feature_stores import BaseFeatureStore
    from zenml.image_builders import BaseImageBuilder
    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)

    return Stack(
        id=id,
        name=name,
        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,
    )

from_model(stack_model) 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.

Source code in src/zenml/stack/stack.py

@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.
    """
    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
    component_models = pagination_utils.depaginate(
        Client().list_stack_components,
        stack_id=stack_model.id,
        hydrate=True,
    )

    stack_components = {
        model.type: StackComponent.from_model(model)
        for model in component_models
    }
    stack = Stack.from_components(
        id=stack_model.id,
        name=stack_model.name,
        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_docker_builds(deployment)

Gets the Docker builds required for the stack.

Parameters:

Name	Type	Description	Default
deployment	PipelineDeploymentBase	The pipeline deployment for which to get the builds.	required

Returns:

Type	Description
List[BuildConfiguration]	The required Docker builds.

Source code in src/zenml/stack/stack.py

def get_docker_builds(
    self, deployment: "PipelineDeploymentBase"
) -> List["BuildConfiguration"]:
    """Gets the Docker builds required for the stack.

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

    Returns:
        The required Docker builds.
    """
    return list(
        itertools.chain.from_iterable(
            component.get_docker_builds(deployment=deployment)
            for component in self.components.values()
        )
    )

get_pipeline_run_metadata(run_id)

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

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.components.values():
        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)

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

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
    ).values():
        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_deployment(deployment)

Prepares the stack for a pipeline deployment.

This method is called before a pipeline is deployed.

Parameters:

Name	Type	Description	Default
deployment	PipelineDeploymentResponse	The pipeline deployment	required

Raises:

Type	Description
RuntimeError	If trying to deploy a pipeline that requires a remote ZenML server with a local one.

Source code in src/zenml/stack/stack.py

def prepare_pipeline_deployment(
    self, deployment: "PipelineDeploymentResponse"
) -> None:
    """Prepares the stack for a pipeline deployment.

    This method is called before a pipeline is deployed.

    Args:
        deployment: The pipeline deployment

    Raises:
        RuntimeError: If trying to deploy 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."
        )

    for component in self.components.values():
        component.prepare_pipeline_deployment(
            deployment=deployment, stack=self
        )

prepare_step_run(info)

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

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
    ).values():
        component.prepare_step_run(info=info)

requirements(exclude_components=None)

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

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.components.values()
        if component.type not in exclude_components
    ]
    return set.union(*requirements) if requirements else set()

validate(fail_if_secrets_missing=False)

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

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()
    for component in self.components.values():
        if component.validator:
            component.validator.validate(stack=self)

    self._validate_secrets(raise_exception=fail_if_secrets_missing)

validate_image_builder()

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

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.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(),
            user=Client().active_user.id,
            workspace=Client().active_workspace.id,
            created=now,
            updated=now,
        )

        self._image_builder = image_builder

StackComponent

Abstract StackComponent class for all components of a ZenML stack.

Source code in src/zenml/stack/stack_component.py

class StackComponent:
    """Abstract StackComponent class for all components of a ZenML stack."""

    def __init__(
        self,
        name: str,
        id: UUID,
        config: StackComponentConfig,
        flavor: str,
        type: StackComponentType,
        user: Optional[UUID],
        workspace: UUID,
        created: datetime,
        updated: datetime,
        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.
            workspace: The ID of the workspace the component belongs to.
            created: The creation time of the component.
            updated: The last update time of the 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.workspace = workspace
        self.created = created
        self.updated = updated
        self.labels = labels
        self.connector_requirements = connector_requirements
        self.connector = connector
        self.connector_resource_id = connector_resource_id
        self._connector_instance: Optional[ServiceConnector] = None

    @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)

        if component_model.user is not None:
            user_id = component_model.user.id
        else:
            user_id = None

        try:
            return flavor.implementation_class(
                user=user_id,
                workspace=component_model.workspace.id,
                name=component_model.name,
                id=component_model.id,
                config=configuration,
                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(
                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

    @property
    def config(self) -> StackComponentConfig:
        """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:
            The configuration of the stack component.
        """
        return self._config

    @property
    def settings_class(self) -> Optional[Type["BaseSettings"]]:
        """Class specifying available settings for this component.

        Returns:
            Optional settings class.
        """
        return None

    def get_settings(
        self,
        container: Union[
            "Step",
            "StepRunResponse",
            "StepRunInfo",
            "PipelineDeploymentBase",
            "PipelineDeploymentResponse",
            "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 `PipelineDeployment` 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."
            )

        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
        )

        if key in all_settings:
            return self.settings_class.model_validate(dict(all_settings[key]))
        else:
            return self.settings_class()

    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()

    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

    @property
    def log_file(self) -> Optional[str]:
        """Optional path to a log file for the stack component.

        Returns:
            Optional path to a log file for the stack component.
        """
        # TODO [ENG-136]: Add support for multiple log files for a stack
        #  component. E.g. let each component return a generator that yields
        #  logs instead of specifying a single file path.
        return None

    @property
    def requirements(self) -> Set[str]:
        """Set of PyPI requirements for the component.

        Returns:
            A set of PyPI requirements for the component.
        """
        from zenml.integrations.utils import get_requirements_for_module

        return set(get_requirements_for_module(self.__module__))

    @property
    def apt_packages(self) -> List[str]:
        """List of APT package requirements for the component.

        Returns:
            A list of APT package requirements for the component.
        """
        from zenml.integrations.utils import get_integration_for_module

        integration = get_integration_for_module(self.__module__)
        return integration.APT_PACKAGES if integration else []

    @property
    def local_path(self) -> Optional[str]:
        """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:

        ```python
        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:
            A path to a local directory used by the component to store
            persistent information.
        """
        return None

    def get_docker_builds(
        self, deployment: "PipelineDeploymentBase"
    ) -> List["BuildConfiguration"]:
        """Gets the Docker builds required for the component.

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

        Returns:
            The required Docker builds.
        """
        return []

    def prepare_pipeline_deployment(
        self,
        deployment: "PipelineDeploymentResponse",
        stack: "Stack",
    ) -> None:
        """Prepares deploying the pipeline.

        This method gets called immediately before a pipeline is deployed.
        Subclasses should override it if they require runtime configuration
        options or if they need to run code before the pipeline deployment.

        Args:
            deployment: The pipeline deployment configuration.
            stack: The stack on which the pipeline will be deployed.
        """

    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 {}

    def prepare_step_run(self, info: "StepRunInfo") -> None:
        """Prepares running a step.

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

    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 {}

    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.
        """

    @property
    def post_registration_message(self) -> Optional[str]:
        """Optional message printed after the stack component is registered.

        Returns:
            An optional message.
        """
        return None

    @property
    def validator(self) -> Optional["StackValidator"]:
        """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:
            An optional `StackValidator` instance.
        """
        return None

    def cleanup(self) -> None:
        """Cleans up the component after it has been used."""
        pass

    def __repr__(self) -> str:
        """String representation of the stack component.

        Returns:
            A string representation of the stack component.
        """
        attribute_representation = ", ".join(
            f"{key}={value}" for key, value in self.config.model_dump().items()
        )
        return (
            f"{self.__class__.__qualname__}(type={self.type}, "
            f"flavor={self.flavor}, {attribute_representation})"
        )

    def __str__(self) -> str:
        """String representation of the stack component.

        Returns:
            A string representation of the stack component.
        """
        return self.__repr__()

apt_packages property

List of APT package requirements for the component.

Returns:

Type	Description
List[str]	A list of APT package requirements for the component.

config 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 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 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 property

Optional message printed after the stack component is registered.

Returns:

Type	Description
Optional[str]	An optional message.

requirements property

Set of PyPI requirements for the component.

Returns:

Type	Description
Set[str]	A set of PyPI requirements for the component.

settings_class property

Class specifying available settings for this component.

Returns:

Type	Description
Optional[Type[BaseSettings]]	Optional settings class.

validator 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.

__init__(name, id, config, flavor, type, user, workspace, created, updated, labels=None, connector_requirements=None, connector=None, connector_resource_id=None, *args, **kwargs)

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
workspace	UUID	The ID of the workspace the component belongs to.	required
created	datetime	The creation time of the component.	required
updated	datetime	The last update time of the component.	required
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

def __init__(
    self,
    name: str,
    id: UUID,
    config: StackComponentConfig,
    flavor: str,
    type: StackComponentType,
    user: Optional[UUID],
    workspace: UUID,
    created: datetime,
    updated: datetime,
    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.
        workspace: The ID of the workspace the component belongs to.
        created: The creation time of the component.
        updated: The last update time of the 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.workspace = workspace
    self.created = created
    self.updated = updated
    self.labels = labels
    self.connector_requirements = connector_requirements
    self.connector = connector
    self.connector_resource_id = connector_resource_id
    self._connector_instance: Optional[ServiceConnector] = None

__repr__()

String representation of the stack component.

Returns:

Type	Description
str	A string representation of the stack component.

Source code in src/zenml/stack/stack_component.py

def __repr__(self) -> str:
    """String representation of the stack component.

    Returns:
        A string representation of the stack component.
    """
    attribute_representation = ", ".join(
        f"{key}={value}" for key, value in self.config.model_dump().items()
    )
    return (
        f"{self.__class__.__qualname__}(type={self.type}, "
        f"flavor={self.flavor}, {attribute_representation})"
    )

__str__()

String representation of the stack component.

Returns:

Type	Description
str	A string representation of the stack component.

Source code in src/zenml/stack/stack_component.py

def __str__(self) -> str:
    """String representation of the stack component.

    Returns:
        A string representation of the stack component.
    """
    return self.__repr__()

cleanup()

Cleans up the component after it has been used.

Source code in src/zenml/stack/stack_component.py

def cleanup(self) -> None:
    """Cleans up the component after it has been used."""
    pass

cleanup_step_run(info, step_failed)

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

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()

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

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) 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

@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)

    if component_model.user is not None:
        user_id = component_model.user.id
    else:
        user_id = None

    try:
        return flavor.implementation_class(
            user=user_id,
            workspace=component_model.workspace.id,
            name=component_model.name,
            id=component_model.id,
            config=configuration,
            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(
            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()

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

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(deployment)

Gets the Docker builds required for the component.

Parameters:

Name	Type	Description	Default
deployment	PipelineDeploymentBase	The pipeline deployment 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

def get_docker_builds(
    self, deployment: "PipelineDeploymentBase"
) -> List["BuildConfiguration"]:
    """Gets the Docker builds required for the component.

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

    Returns:
        The required Docker builds.
    """
    return []

get_pipeline_run_metadata(run_id)

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

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)

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, PipelineDeploymentBase, PipelineDeploymentResponse, PipelineRunResponse]	The Step, StepRunInfo or PipelineDeployment 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

def get_settings(
    self,
    container: Union[
        "Step",
        "StepRunResponse",
        "StepRunInfo",
        "PipelineDeploymentBase",
        "PipelineDeploymentResponse",
        "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 `PipelineDeployment` 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."
        )

    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
    )

    if key in all_settings:
        return self.settings_class.model_validate(dict(all_settings[key]))
    else:
        return self.settings_class()

get_step_run_metadata(info)

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

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_pipeline_deployment(deployment, stack)

Prepares deploying the pipeline.

This method gets called immediately before a pipeline is deployed. Subclasses should override it if they require runtime configuration options or if they need to run code before the pipeline deployment.

Parameters:

Name	Type	Description	Default
deployment	PipelineDeploymentResponse	The pipeline deployment configuration.	required
stack	Stack	The stack on which the pipeline will be deployed.	required

Source code in src/zenml/stack/stack_component.py

def prepare_pipeline_deployment(
    self,
    deployment: "PipelineDeploymentResponse",
    stack: "Stack",
) -> None:
    """Prepares deploying the pipeline.

    This method gets called immediately before a pipeline is deployed.
    Subclasses should override it if they require runtime configuration
    options or if they need to run code before the pipeline deployment.

    Args:
        deployment: The pipeline deployment configuration.
        stack: The stack on which the pipeline will be deployed.
    """

prepare_step_run(info)

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

def prepare_step_run(self, info: "StepRunInfo") -> None:
    """Prepares running a step.

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

StackComponentConfig

Bases: BaseModel, ABC

Base class for all ZenML stack component configs.

Source code in src/zenml/stack/stack_component.py

class StackComponentConfig(BaseModel, ABC):
    """Base class for all ZenML stack component configs."""

    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/getting-started/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)

    @property
    def required_secrets(self) -> Set[secret_utils.SecretReference]:
        """All required secrets for this stack component.

        Returns:
            The required secrets of this stack component.
        """
        return {
            secret_utils.parse_secret_reference(v)
            for v in self.model_dump().values()
            if secret_utils.is_secret_reference(v)
        }

    @property
    def is_remote(self) -> bool:
        """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:
            True if this config is for a remote component, False otherwise.
        """
        return False

    @property
    def is_valid(self) -> bool:
        """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:
            True if the stack component config is valid, False otherwise.
        """
        return True

    @property
    def is_local(self) -> bool:
        """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:
            True if this config is for a local component, False otherwise.
        """
        return False

    def __custom_getattribute__(self, key: str) -> Any:
        """Returns the (potentially resolved) attribute value for the given key.

        An attribute value may be either specified directly, or as a secret
        reference. In case of a secret reference, this method resolves the
        reference and returns the secret value instead.

        Args:
            key: The key for which to get the attribute value.

        Raises:
            KeyError: If the secret or secret key don't exist.

        Returns:
            The (potentially resolved) attribute value.
        """
        from zenml.client import Client

        value = super().__getattribute__(key)

        if not secret_utils.is_secret_reference(value):
            return value

        secret_ref = secret_utils.parse_secret_reference(value)

        # Try to resolve the secret using the secret store
        try:
            secret = Client().get_secret_by_name_and_scope(
                name=secret_ref.name,
            )
        except (KeyError, NotImplementedError):
            raise KeyError(
                f"Failed to resolve secret reference for attribute {key} "
                f"of stack component `{self}`: The secret "
                f"{secret_ref.name} does not exist."
            )

        if secret_ref.key not in secret.values:
            raise KeyError(
                f"Failed to resolve secret reference for attribute {key} "
                f"of stack component `{self}`. "
                f"The secret {secret_ref.name} does not contain a value "
                f"for key {secret_ref.key}. Available keys: "
                f"{set(secret.values.keys())}."
            )

        return secret.secret_values[secret_ref.key]

    def _is_part_of_active_stack(self) -> bool:
        """Checks if this config belongs to a component in the active stack.

        Returns:
            True if this config belongs to a component in the active stack,
            False otherwise.
        """
        from zenml.client import Client

        for component in Client().active_stack.components.values():
            if component.config == self:
                return True
        return False

    if not TYPE_CHECKING:
        # When defining __getattribute__, mypy allows accessing non-existent
        # attributes without failing
        # (see https://github.com/python/mypy/issues/13319).
        __getattribute__ = __custom_getattribute__

    @model_validator(mode="before")
    @classmethod
    @pydantic_utils.before_validator_handler
    def _convert_json_strings(cls, data: Dict[str, Any]) -> Dict[str, Any]:
        """Converts potential JSON strings.

        Args:
            data: The model data.

        Returns:
            The potentially converted data.

        Raises:
            ValueError: If any of the values is an invalid JSON string.
        """
        for key, field in cls.model_fields.items():
            if not field.annotation:
                continue

            value = data.get(key, None)

            if isinstance(value, str):
                if typing_utils.is_optional(field.annotation):
                    args = list(typing_utils.get_args(field.annotation))
                    if str in args:
                        # Don't do any type coercion in case str is in the
                        # possible types of the field
                        continue

                    # Remove `NoneType` from the arguments
                    NoneType = type(None)
                    if NoneType in args:
                        args.remove(NoneType)

                    # We just choose the first arg and match against this
                    annotation = args[0]
                else:
                    annotation = field.annotation

                if typing_utils.get_origin(annotation) in {
                    dict,
                    list,
                    Mapping,
                    Sequence,
                }:
                    try:
                        data[key] = json.loads(value)
                    except json.JSONDecodeError as e:
                        raise ValueError(
                            f"Invalid json string '{value}'"
                        ) from e
                elif isclass(annotation) and issubclass(annotation, BaseModel):
                    data[key] = annotation.model_validate_json(
                        value
                    ).model_dump()

        return data

    model_config = ConfigDict(
        # public attributes are immutable
        frozen=True,
        # prevent extra attributes during model initialization
        extra="forbid",
    )

is_local 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 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 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 property

All required secrets for this stack component.

Returns:

Type	Description
Set[SecretReference]	The required secrets of this stack component.

__custom_getattribute__(key)

Returns the (potentially resolved) attribute value for the given key.

An attribute value may be either specified directly, or as a secret reference. In case of a secret reference, this method resolves the reference and returns the secret value instead.

Parameters:

Name	Type	Description	Default
key	str	The key for which to get the attribute value.	required

Raises:

Type	Description
KeyError	If the secret or secret key don't exist.

Returns:

Type	Description
Any	The (potentially resolved) attribute value.

Source code in src/zenml/stack/stack_component.py

def __custom_getattribute__(self, key: str) -> Any:
    """Returns the (potentially resolved) attribute value for the given key.

    An attribute value may be either specified directly, or as a secret
    reference. In case of a secret reference, this method resolves the
    reference and returns the secret value instead.

    Args:
        key: The key for which to get the attribute value.

    Raises:
        KeyError: If the secret or secret key don't exist.

    Returns:
        The (potentially resolved) attribute value.
    """
    from zenml.client import Client

    value = super().__getattribute__(key)

    if not secret_utils.is_secret_reference(value):
        return value

    secret_ref = secret_utils.parse_secret_reference(value)

    # Try to resolve the secret using the secret store
    try:
        secret = Client().get_secret_by_name_and_scope(
            name=secret_ref.name,
        )
    except (KeyError, NotImplementedError):
        raise KeyError(
            f"Failed to resolve secret reference for attribute {key} "
            f"of stack component `{self}`: The secret "
            f"{secret_ref.name} does not exist."
        )

    if secret_ref.key not in secret.values:
        raise KeyError(
            f"Failed to resolve secret reference for attribute {key} "
            f"of stack component `{self}`. "
            f"The secret {secret_ref.name} does not contain a value "
            f"for key {secret_ref.key}. Available keys: "
            f"{set(secret.values.keys())}."
        )

    return secret.secret_values[secret_ref.key]

__init__(warn_about_plain_text_secrets=False, **kwargs)

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

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/getting-started/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)

StackValidator

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.

Source code in src/zenml/stack/stack_validator.py

class StackValidator:
    """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.
    """

    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

    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.
        """
        missing_components = self._required_components - set(stack.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}"
                )

__init__(required_components=None, custom_validation_function=None)

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

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

validate(stack)

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

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.
    """
    missing_components = self._required_components - set(stack.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}"
            )