Config

zenml.config

The config module contains classes and functions that manage user-specific configuration.

ZenML's configuration is stored in a file called config.yaml, located on the user's directory for configuration files. (The exact location differs from operating system to operating system.)

The GlobalConfiguration class is the main class in this module. It provides a Pydantic configuration object that is used to store and retrieve configuration. This GlobalConfiguration object handles the serialization and deserialization of the configuration options that are stored in the file in order to persist the configuration across sessions.

Attributes

__all__ = ['DockerSettings', 'ResourceSettings', 'StepRetryConfig'] module-attribute

Classes

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

Bases: BaseSettings

Settings for building Docker images to run ZenML pipelines.

Build process:

• No dockerfile specified: If any of the options regarding requirements, environment variables or copying files require us to build an image, ZenML will build this image. Otherwise, the parent_image will be used to run the pipeline.
• dockerfile specified: ZenML will first build an image based on the specified Dockerfile. If any of the options regarding requirements, environment variables or copying files require an additional image built on top of that, ZenML will build a second image. If not, the image build from the specified Dockerfile will be used to run the pipeline.

Requirements installation order:

Depending on the configuration of this object, requirements will be installed in the following order (each step optional): - The packages installed in your local python environment - The packages required by the stack unless this is disabled by setting install_stack_requirements=False. - The packages specified via the required_integrations - The packages specified via the requirements attribute

Attributes:

Name	Type	Description
parent_image	Optional[str]	Full name of the Docker image that should be used as the parent for the image that will be built. Defaults to a ZenML image built for the active Python and ZenML version. Additional notes: * If you specify a custom image here, you need to make sure it has ZenML installed. * If this is a non-local image, the environment which is running the pipeline and building the Docker image needs to be able to pull this image. * If a custom dockerfile is specified for this settings object, this parent image will be ignored.
dockerfile	Optional[str]	Path to a custom Dockerfile that should be built. Depending on the other values you specify in this object, the resulting image will be used directly to run your pipeline or ZenML will use it as a parent image to build on top of. See the general docstring of this class for more information. Additional notes: * If you specify this, the parent_image attribute will be ignored. * If you specify this, the image built from this Dockerfile needs to have ZenML installed.
build_context_root	Optional[str]	Build context root for the Docker build, only used when the dockerfile attribute is set. If this is left empty, the build context will only contain the Dockerfile.
parent_image_build_config	Optional[DockerBuildConfig]	Configuration for the parent image build.
skip_build	bool	If set to True, the parent image will be used directly to run the steps of your pipeline.
prevent_build_reuse	bool	Prevent the reuse of an existing build.
target_repository	Optional[str]	Name of the Docker repository to which the image should be pushed. This repository will be appended to the registry URI of the container registry of your stack and should therefore not include any registry. If not specified, the default repository name configured in the container registry stack component settings will be used.
python_package_installer	PythonPackageInstaller	The package installer to use for python packages.
python_package_installer_args	Dict[str, Any]	Arguments to pass to the python package installer.
replicate_local_python_environment	Optional[Union[List[str], PythonEnvironmentExportMethod]]	If not None, ZenML will use the specified method to generate a requirements file that replicates the packages installed in the currently running python environment. This requirements file will then be installed in the Docker image.
requirements	Union[None, str, List[str]]	Path to a requirements file or a list of required pip packages. During the image build, these requirements will be installed using pip. If you need to use a different tool to resolve and/or install your packages, please use a custom parent image or specify a custom dockerfile.
required_integrations	List[str]	List of ZenML integrations that should be installed. All requirements for the specified integrations will be installed inside the Docker image.
required_hub_plugins	List[str]	DEPRECATED/UNUSED.
install_stack_requirements	bool	If True, ZenML will automatically detect if components of your active stack are part of a ZenML integration and install the corresponding requirements and apt packages. If you set this to False or use custom components in your stack, you need to make sure these get installed by specifying them in the requirements and apt_packages attributes.
apt_packages	List[str]	APT packages to install inside the Docker image.
environment	Dict[str, Any]	Dictionary of environment variables to set inside the Docker image.
build_config	Optional[DockerBuildConfig]	Configuration for the main image build.
user	Optional[str]	If not None, will set the user, make it owner of the /app directory which contains all the user code and run the container entrypoint as this user.
allow_including_files_in_images	bool	If True, code can be included in the Docker images if code download from a code repository or artifact store is disabled or not possible.
allow_download_from_code_repository	bool	If True, code can be downloaded from a code repository if possible.
allow_download_from_artifact_store	bool	If True, code can be downloaded from the artifact store.
build_options	Dict[str, Any]	DEPRECATED, use parent_image_build_config.build_options instead.
dockerignore	Optional[str]	DEPRECATED, use build_config.dockerignore instead.
copy_files	bool	DEPRECATED/UNUSED.
copy_global_config	bool	DEPRECATED/UNUSED.
source_files	Optional[str]	DEPRECATED. Use allow_including_files_in_images, allow_download_from_code_repository and allow_download_from_artifact_store instead.

Source code in src/zenml/config/secret_reference_mixin.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references are only passed for valid fields.

    This method ensures that secret references are not passed for fields
    that explicitly prevent them or require pydantic validation.

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

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            or an attribute which explicitly disallows secret references
            is 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}`. 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 values with secrets "
                    "here: https://docs.zenml.io/getting-started/deploying-zenml/secret-management"
                )
            continue

        if secret_utils.is_clear_text_field(field):
            raise ValueError(
                f"Passing the `{key}` attribute as a secret reference is "
                "not allowed."
            )

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

    super().__init__(**kwargs)

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

Bases: BaseSettings

Hardware resource settings.

Attributes:

Name	Type	Description
cpu_count	Optional[PositiveFloat]	The amount of CPU cores that should be configured.
gpu_count	Optional[NonNegativeInt]	The amount of GPUs that should be configured.
memory	Optional[str]	The amount of memory that should be configured.

Source code in src/zenml/config/secret_reference_mixin.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references are only passed for valid fields.

    This method ensures that secret references are not passed for fields
    that explicitly prevent them or require pydantic validation.

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

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            or an attribute which explicitly disallows secret references
            is 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}`. 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 values with secrets "
                    "here: https://docs.zenml.io/getting-started/deploying-zenml/secret-management"
                )
            continue

        if secret_utils.is_clear_text_field(field):
            raise ValueError(
                f"Passing the `{key}` attribute as a secret reference is "
                "not allowed."
            )

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

    super().__init__(**kwargs)

Attributes

empty: bool property

Returns if this object is "empty" (=no values configured) or not.

Returns:

Type	Description
bool	True if no values were configured, False otherwise.

Functions

get_memory(unit: Union[str, ByteUnit] = ByteUnit.GB) -> Optional[float]

Gets the memory configuration in a specific unit.

Parameters:

Name	Type	Description	Default
unit	Union[str, ByteUnit]	The unit to which the memory should be converted.	GB

Raises:

Type	Description
ValueError	If the memory string is invalid.

Returns:

Type	Description
Optional[float]	The memory configuration converted to the requested unit, or None
Optional[float]	if no memory was configured.

Source code in src/zenml/config/resource_settings.py

def get_memory(
    self, unit: Union[str, ByteUnit] = ByteUnit.GB
) -> Optional[float]:
    """Gets the memory configuration in a specific unit.

    Args:
        unit: The unit to which the memory should be converted.

    Raises:
        ValueError: If the memory string is invalid.

    Returns:
        The memory configuration converted to the requested unit, or None
        if no memory was configured.
    """
    if not self.memory:
        return None

    if isinstance(unit, str):
        unit = ByteUnit(unit)

    memory = self.memory
    for memory_unit in ByteUnit:
        if memory.endswith(memory_unit.value):
            memory_value = int(memory[: -len(memory_unit.value)])
            return memory_value * memory_unit.byte_value / unit.byte_value
    else:
        # Should never happen due to the regex validation
        raise ValueError(f"Unable to parse memory unit from '{memory}'.")

StepRetryConfig

Bases: StrictBaseModel

Retry configuration for a step.

Delay is an integer (specified in seconds).

Modules

base_settings

Base class for all ZenML settings.

Classes

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

Bases: SecretReferenceMixin

Base class for settings.

The LEVEL class variable defines on which level the settings can be specified. By default, subclasses can be defined on both pipelines and steps.

Source code in src/zenml/config/secret_reference_mixin.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references are only passed for valid fields.

    This method ensures that secret references are not passed for fields
    that explicitly prevent them or require pydantic validation.

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

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            or an attribute which explicitly disallows secret references
            is 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}`. 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 values with secrets "
                    "here: https://docs.zenml.io/getting-started/deploying-zenml/secret-management"
                )
            continue

        if secret_utils.is_clear_text_field(field):
            raise ValueError(
                f"Passing the `{key}` attribute as a secret reference is "
                "not allowed."
            )

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

    super().__init__(**kwargs)

ConfigurationLevel

Bases: IntFlag

Settings configuration level.

Bit flag that can be used to specify where a BaseSettings subclass can be specified.

build_configuration

Build configuration class.

Classes

BuildConfiguration

Bases: BaseModel

Configuration of Docker builds.

Attributes:

Name	Type	Description
key	str	The key to store the build.
settings	DockerSettings	Settings for the build.
step_name	Optional[str]	Name of the step for which this image will be built.
entrypoint	Optional[str]	Optional entrypoint for the image.
extra_files	Dict[str, str]	Extra files to include in the Docker image.

Functions

compute_settings_checksum(stack: Stack, code_repository: Optional[BaseCodeRepository] = None) -> str

Checksum for all build settings.

Parameters:

Name	Type	Description	Default
stack	Stack	The stack for which to compute the checksum. This is needed to gather the stack integration requirements in case the Docker settings specify to install them.	required
code_repository	Optional[BaseCodeRepository]	Optional code repository that will be used to download files inside the image.	None

Returns:

Type	Description
str	The checksum.

Source code in src/zenml/config/build_configuration.py

def compute_settings_checksum(
    self,
    stack: "Stack",
    code_repository: Optional["BaseCodeRepository"] = None,
) -> str:
    """Checksum for all build settings.

    Args:
        stack: The stack for which to compute the checksum. This is needed
            to gather the stack integration requirements in case the
            Docker settings specify to install them.
        code_repository: Optional code repository that will be used to
            download files inside the image.

    Returns:
        The checksum.
    """
    hash_ = hashlib.md5()  # nosec
    settings_json = json.dumps(
        self.settings.model_dump(
            mode="json", exclude={"prevent_build_reuse"}
        ),
        sort_keys=True,
        default=json_utils.pydantic_encoder,
    )
    hash_.update(settings_json.encode())
    if self.entrypoint:
        hash_.update(self.entrypoint.encode())

    for destination, source in self.extra_files.items():
        hash_.update(destination.encode())
        hash_.update(source.encode())

    from zenml.utils.pipeline_docker_image_builder import (
        PipelineDockerImageBuilder,
    )

    pass_code_repo = self.should_download_files_from_code_repository(
        code_repository=code_repository
    )
    requirements_files = (
        PipelineDockerImageBuilder.gather_requirements_files(
            docker_settings=self.settings,
            stack=stack,
            code_repository=code_repository if pass_code_repo else None,
            log=False,
        )
    )
    for _, requirements, _ in requirements_files:
        hash_.update(requirements.encode())

    return hash_.hexdigest()

should_download_files(code_repository: Optional[BaseCodeRepository]) -> bool

Whether files should be downloaded in the image.

Parameters:

Name	Type	Description	Default
code_repository	Optional[BaseCodeRepository]	Code repository that can be used to download files inside the image.	required

Returns:

Type	Description
bool	Whether files should be downloaded in the image.

Source code in src/zenml/config/build_configuration.py

def should_download_files(
    self,
    code_repository: Optional["BaseCodeRepository"],
) -> bool:
    """Whether files should be downloaded in the image.

    Args:
        code_repository: Code repository that can be used to download files
            inside the image.

    Returns:
        Whether files should be downloaded in the image.
    """
    if self.should_download_files_from_code_repository(
        code_repository=code_repository
    ):
        return True

    if self.settings.allow_download_from_artifact_store:
        return True

    return False

should_download_files_from_code_repository(code_repository: Optional[BaseCodeRepository]) -> bool

Whether files should be downloaded from the code repository.

Parameters:

Name	Type	Description	Default
code_repository	Optional[BaseCodeRepository]	Code repository that can be used to download files inside the image.	required

Returns:

Type	Description
bool	Whether files should be downloaded from the code repository.

Source code in src/zenml/config/build_configuration.py

def should_download_files_from_code_repository(
    self,
    code_repository: Optional["BaseCodeRepository"],
) -> bool:
    """Whether files should be downloaded from the code repository.

    Args:
        code_repository: Code repository that can be used to download files
            inside the image.

    Returns:
        Whether files should be downloaded from the code repository.
    """
    if (
        code_repository
        and self.settings.allow_download_from_code_repository
    ):
        return True

    return False

should_include_files(code_repository: Optional[BaseCodeRepository]) -> bool

Whether files should be included in the image.

Parameters:

Name	Type	Description	Default
code_repository	Optional[BaseCodeRepository]	Code repository that can be used to download files inside the image.	required

Returns:

Type	Description
bool	Whether files should be included in the image.

Source code in src/zenml/config/build_configuration.py

def should_include_files(
    self,
    code_repository: Optional["BaseCodeRepository"],
) -> bool:
    """Whether files should be included in the image.

    Args:
        code_repository: Code repository that can be used to download files
            inside the image.

    Returns:
        Whether files should be included in the image.
    """
    if self.should_download_files(code_repository=code_repository):
        return False

    return self.settings.allow_including_files_in_images

Modules

compiler

Class for compiling ZenML pipelines into a serializable format.

Classes

Compiler

Compiles ZenML pipelines to serializable representations.

Functions

compile(pipeline: Pipeline, stack: Stack, run_configuration: PipelineRunConfiguration) -> PipelineDeploymentBase

Compiles a ZenML pipeline to a serializable representation.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	The pipeline to compile.	required
stack	Stack	The stack on which the pipeline will run.	required
run_configuration	PipelineRunConfiguration	The run configuration for this pipeline.	required

Returns:

Type	Description
PipelineDeploymentBase	The compiled pipeline deployment.

Source code in src/zenml/config/compiler.py

def compile(
    self,
    pipeline: "Pipeline",
    stack: "Stack",
    run_configuration: PipelineRunConfiguration,
) -> PipelineDeploymentBase:
    """Compiles a ZenML pipeline to a serializable representation.

    Args:
        pipeline: The pipeline to compile.
        stack: The stack on which the pipeline will run.
        run_configuration: The run configuration for this pipeline.

    Returns:
        The compiled pipeline deployment.
    """
    logger.debug("Compiling pipeline `%s`.", pipeline.name)
    # Copy the pipeline before we apply any run-level configurations, so
    # we don't mess with the pipeline object/step objects in any way
    pipeline = copy.deepcopy(pipeline)
    self._apply_run_configuration(
        pipeline=pipeline, config=run_configuration
    )
    convert_component_shortcut_settings_keys(
        pipeline.configuration.settings, stack=stack
    )

    self._apply_stack_default_settings(pipeline=pipeline, stack=stack)
    if run_configuration.run_name:
        self._verify_run_name(
            run_configuration.run_name,
            pipeline.configuration.substitutions,
        )

    pipeline_settings = self._filter_and_validate_settings(
        settings=pipeline.configuration.settings,
        configuration_level=ConfigurationLevel.PIPELINE,
        stack=stack,
    )
    with pipeline.__suppress_configure_warnings__():
        pipeline.configure(settings=pipeline_settings, merge=False)

    settings_to_passdown = {
        key: settings
        for key, settings in pipeline_settings.items()
        if ConfigurationLevel.STEP in settings.LEVEL
    }

    steps = {
        invocation_id: self._compile_step_invocation(
            invocation=invocation,
            pipeline_settings=settings_to_passdown,
            pipeline_extra=pipeline.configuration.extra,
            stack=stack,
            step_config=run_configuration.steps.get(invocation_id),
            pipeline_failure_hook_source=pipeline.configuration.failure_hook_source,
            pipeline_success_hook_source=pipeline.configuration.success_hook_source,
        )
        for invocation_id, invocation in self._get_sorted_invocations(
            pipeline=pipeline
        )
    }

    self._ensure_required_stack_components_exist(stack=stack, steps=steps)

    run_name = run_configuration.run_name or get_default_run_name(
        pipeline_name=pipeline.name
    )

    client_version, server_version = get_zenml_versions()

    step_specs = [step.spec for step in steps.values()]
    pipeline_spec = self._compute_pipeline_spec(
        pipeline=pipeline, step_specs=step_specs
    )

    deployment = PipelineDeploymentBase(
        run_name_template=run_name,
        pipeline_configuration=pipeline.configuration,
        step_configurations=steps,
        client_environment=get_run_environment_dict(),
        client_version=client_version,
        server_version=server_version,
        pipeline_version_hash=pipeline._compute_unique_identifier(
            pipeline_spec=pipeline_spec
        ),
        pipeline_spec=pipeline_spec,
    )

    logger.debug("Compiled pipeline deployment: %s", deployment)

    return deployment

compile_spec(pipeline: Pipeline) -> PipelineSpec

Compiles a ZenML pipeline to a pipeline spec.

This method can be used when a pipeline spec is needed but the full deployment including stack information is not required.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	The pipeline to compile.	required

Returns:

Type	Description
PipelineSpec	The compiled pipeline spec.

Source code in src/zenml/config/compiler.py

def compile_spec(self, pipeline: "Pipeline") -> PipelineSpec:
    """Compiles a ZenML pipeline to a pipeline spec.

    This method can be used when a pipeline spec is needed but the full
    deployment including stack information is not required.

    Args:
        pipeline: The pipeline to compile.

    Returns:
        The compiled pipeline spec.
    """
    logger.debug(
        "Compiling pipeline spec for pipeline `%s`.", pipeline.name
    )
    # Copy the pipeline before we connect the steps, so we don't mess with
    # the pipeline object/step objects in any way
    pipeline = copy.deepcopy(pipeline)

    invocations = [
        self._get_step_spec(invocation=invocation)
        for _, invocation in self._get_sorted_invocations(
            pipeline=pipeline
        )
    ]

    pipeline_spec = self._compute_pipeline_spec(
        pipeline=pipeline, step_specs=invocations
    )
    logger.debug("Compiled pipeline spec: %s", pipeline_spec)
    return pipeline_spec

Functions

convert_component_shortcut_settings_keys(settings: Dict[str, BaseSettings], stack: Stack) -> None

Convert component shortcut settings keys.

Parameters:

Name	Type	Description	Default
settings	Dict[str, BaseSettings]	Dictionary of settings.	required
stack	Stack	The stack that the pipeline will run on.	required

Raises:

Type	Description
ValueError	If stack component settings were defined both using the full and the shortcut key.

Source code in src/zenml/config/compiler.py

def convert_component_shortcut_settings_keys(
    settings: Dict[str, "BaseSettings"], stack: "Stack"
) -> None:
    """Convert component shortcut settings keys.

    Args:
        settings: Dictionary of settings.
        stack: The stack that the pipeline will run on.

    Raises:
        ValueError: If stack component settings were defined both using the
            full and the shortcut key.
    """
    for component in stack.components.values():
        shortcut_key = str(component.type)
        if component_settings := settings.pop(shortcut_key, None):
            key = settings_utils.get_stack_component_setting_key(component)
            if key in settings:
                raise ValueError(
                    f"Duplicate settings provided for your {shortcut_key} "
                    f"using the keys {shortcut_key} and {key}. Remove settings "
                    "for one of them to fix this error."
                )

            settings[key] = component_settings

get_zenml_versions() -> Tuple[str, str]

Returns the version of ZenML on the client and server side.

Returns:

Type	Description
Tuple[str, str]	the ZenML versions on the client and server side respectively.

Source code in src/zenml/config/compiler.py

def get_zenml_versions() -> Tuple[str, str]:
    """Returns the version of ZenML on the client and server side.

    Returns:
        the ZenML versions on the client and server side respectively.
    """
    from zenml.client import Client

    client = Client()
    server_version = client.zen_store.get_store_info().version

    return __version__, server_version

Modules

constants

ZenML settings constants.

docker_settings

Docker settings.

Classes

DockerBuildConfig

Bases: BaseModel

Configuration for a Docker build.

Attributes:

Name	Type	Description
build_options	Dict[str, Any]	Additional options that will be passed unmodified to the Docker build call when building an image. You can use this to for example specify build args or a target stage. See https://docker-py.readthedocs.io/en/stable/images.html#docker.models.images.ImageCollection.build for a full list of available options.
dockerignore	Optional[str]	Path to a dockerignore file to use when building the Docker image.

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

Bases: BaseSettings

Settings for building Docker images to run ZenML pipelines.

Build process:

• No dockerfile specified: If any of the options regarding requirements, environment variables or copying files require us to build an image, ZenML will build this image. Otherwise, the parent_image will be used to run the pipeline.
• dockerfile specified: ZenML will first build an image based on the specified Dockerfile. If any of the options regarding requirements, environment variables or copying files require an additional image built on top of that, ZenML will build a second image. If not, the image build from the specified Dockerfile will be used to run the pipeline.

Requirements installation order:

Depending on the configuration of this object, requirements will be installed in the following order (each step optional): - The packages installed in your local python environment - The packages required by the stack unless this is disabled by setting install_stack_requirements=False. - The packages specified via the required_integrations - The packages specified via the requirements attribute

Attributes:

Name	Type	Description
parent_image	Optional[str]	Full name of the Docker image that should be used as the parent for the image that will be built. Defaults to a ZenML image built for the active Python and ZenML version. Additional notes: * If you specify a custom image here, you need to make sure it has ZenML installed. * If this is a non-local image, the environment which is running the pipeline and building the Docker image needs to be able to pull this image. * If a custom dockerfile is specified for this settings object, this parent image will be ignored.
dockerfile	Optional[str]	Path to a custom Dockerfile that should be built. Depending on the other values you specify in this object, the resulting image will be used directly to run your pipeline or ZenML will use it as a parent image to build on top of. See the general docstring of this class for more information. Additional notes: * If you specify this, the parent_image attribute will be ignored. * If you specify this, the image built from this Dockerfile needs to have ZenML installed.
build_context_root	Optional[str]	Build context root for the Docker build, only used when the dockerfile attribute is set. If this is left empty, the build context will only contain the Dockerfile.
parent_image_build_config	Optional[DockerBuildConfig]	Configuration for the parent image build.
skip_build	bool	If set to True, the parent image will be used directly to run the steps of your pipeline.
prevent_build_reuse	bool	Prevent the reuse of an existing build.
target_repository	Optional[str]	Name of the Docker repository to which the image should be pushed. This repository will be appended to the registry URI of the container registry of your stack and should therefore not include any registry. If not specified, the default repository name configured in the container registry stack component settings will be used.
python_package_installer	PythonPackageInstaller	The package installer to use for python packages.
python_package_installer_args	Dict[str, Any]	Arguments to pass to the python package installer.
replicate_local_python_environment	Optional[Union[List[str], PythonEnvironmentExportMethod]]	If not None, ZenML will use the specified method to generate a requirements file that replicates the packages installed in the currently running python environment. This requirements file will then be installed in the Docker image.
requirements	Union[None, str, List[str]]	Path to a requirements file or a list of required pip packages. During the image build, these requirements will be installed using pip. If you need to use a different tool to resolve and/or install your packages, please use a custom parent image or specify a custom dockerfile.
required_integrations	List[str]	List of ZenML integrations that should be installed. All requirements for the specified integrations will be installed inside the Docker image.
required_hub_plugins	List[str]	DEPRECATED/UNUSED.
install_stack_requirements	bool	If True, ZenML will automatically detect if components of your active stack are part of a ZenML integration and install the corresponding requirements and apt packages. If you set this to False or use custom components in your stack, you need to make sure these get installed by specifying them in the requirements and apt_packages attributes.
apt_packages	List[str]	APT packages to install inside the Docker image.
environment	Dict[str, Any]	Dictionary of environment variables to set inside the Docker image.
build_config	Optional[DockerBuildConfig]	Configuration for the main image build.
user	Optional[str]	If not None, will set the user, make it owner of the /app directory which contains all the user code and run the container entrypoint as this user.
allow_including_files_in_images	bool	If True, code can be included in the Docker images if code download from a code repository or artifact store is disabled or not possible.
allow_download_from_code_repository	bool	If True, code can be downloaded from a code repository if possible.
allow_download_from_artifact_store	bool	If True, code can be downloaded from the artifact store.
build_options	Dict[str, Any]	DEPRECATED, use parent_image_build_config.build_options instead.
dockerignore	Optional[str]	DEPRECATED, use build_config.dockerignore instead.
copy_files	bool	DEPRECATED/UNUSED.
copy_global_config	bool	DEPRECATED/UNUSED.
source_files	Optional[str]	DEPRECATED. Use allow_including_files_in_images, allow_download_from_code_repository and allow_download_from_artifact_store instead.

Source code in src/zenml/config/secret_reference_mixin.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references are only passed for valid fields.

    This method ensures that secret references are not passed for fields
    that explicitly prevent them or require pydantic validation.

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

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            or an attribute which explicitly disallows secret references
            is 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}`. 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 values with secrets "
                    "here: https://docs.zenml.io/getting-started/deploying-zenml/secret-management"
                )
            continue

        if secret_utils.is_clear_text_field(field):
            raise ValueError(
                f"Passing the `{key}` attribute as a secret reference is "
                "not allowed."
            )

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

    super().__init__(**kwargs)

PythonEnvironmentExportMethod

Bases: Enum

Different methods to export the local Python environment.

Attributes

command: str property

Shell command that outputs local python packages.

The output string must be something that can be interpreted as a requirements file for pip once it's written to a file.

Returns:

Type	Description
str	Shell command.

PythonPackageInstaller

Bases: Enum

Different installers for python packages.

Functions

Modules

global_config

Functionality to support ZenML GlobalConfiguration.

Classes

GlobalConfigMetaClass(*args: Any, **kwargs: Any)

Bases: ModelMetaclass

Global configuration metaclass.

This metaclass is used to enforce a singleton instance of the GlobalConfiguration class with the following additional properties:

• the GlobalConfiguration is initialized automatically on import with the default configuration, if no config file exists yet.
• the GlobalConfiguration undergoes a schema migration if the version of the config file is older than the current version of the ZenML package.
• a default store is set if no store is configured yet.

Initialize a singleton class.

Parameters:

Name	Type	Description	Default
*args	Any	positional arguments	()
**kwargs	Any	keyword arguments	{}

Source code in src/zenml/config/global_config.py

def __init__(cls, *args: Any, **kwargs: Any) -> None:
    """Initialize a singleton class.

    Args:
        *args: positional arguments
        **kwargs: keyword arguments
    """
    super().__init__(*args, **kwargs)
    cls._global_config: Optional["GlobalConfiguration"] = None

Functions

GlobalConfiguration(**data: Any)

Bases: BaseModel

Stores global configuration options.

Configuration options are read from a config file, but can be overwritten by environment variables. See GlobalConfiguration.__getattribute__ for more details.

Attributes:

Name	Type	Description
user_id	UUID	Unique user id.
user_email	Optional[str]	Email address associated with this client.
user_email_opt_in	Optional[bool]	Whether the user has opted in to email communication.
analytics_opt_in	bool	If a user agreed to sending analytics or not.
version	Optional[str]	Version of ZenML that was last used to create or update the global config.
store	Optional[SerializeAsAny[StoreConfiguration]]	Store configuration.
active_stack_id	Optional[UUID]	The ID of the active stack.
active_project_id	Optional[UUID]	The ID of the active project.

Initializes a GlobalConfiguration using values from the config file.

GlobalConfiguration is a singleton class: only one instance can exist. Calling this constructor multiple times will always yield the same instance.

Parameters:

Name	Type	Description	Default
data	Any	Custom configuration options.	{}

Source code in src/zenml/config/global_config.py

def __init__(self, **data: Any) -> None:
    """Initializes a GlobalConfiguration using values from the config file.

    GlobalConfiguration is a singleton class: only one instance can exist.
    Calling this constructor multiple times will always yield the same
    instance.

    Args:
        data: Custom configuration options.
    """
    config_values = self._read_config()
    config_values.update(data)

    super().__init__(**config_values)

    if not fileio.exists(self._config_file):
        self._write_config()

Attributes

config_directory: str property

Directory where the global configuration file is located.

Returns:

Type	Description
str	The directory where the global configuration file is located.

is_initialized: bool property

Check if the global configuration is initialized.

Returns:

Type	Description
bool	True if the global configuration is initialized.

local_stores_path: str property

Path where local stores information is stored.

Returns:

Type	Description
str	The path where local stores information is stored.

store_configuration: StoreConfiguration property

Get the current store configuration.

Returns:

Type	Description
StoreConfiguration	The store configuration.

zen_store: BaseZenStore property

Initialize and/or return the global zen store.

If the store hasn't been initialized yet, it is initialized when this property is first accessed according to the global store configuration.

Returns:

Type	Description
BaseZenStore	The current zen store.

Functions

get_active_project() -> ProjectResponse

Get a model of the active project for the local client.

Returns:

Type	Description
ProjectResponse	The model of the active project.

Source code in src/zenml/config/global_config.py

def get_active_project(self) -> "ProjectResponse":
    """Get a model of the active project for the local client.

    Returns:
        The model of the active project.
    """
    project_id = self.get_active_project_id()

    if self._active_project is not None:
        return self._active_project

    project = self.zen_store.get_project(
        project_name_or_id=project_id,
    )
    return self.set_active_project(project)

get_active_project_id() -> UUID

Get the ID of the active project.

Returns:

Type	Description
UUID	The ID of the active project.

Raises:

Type	Description
RuntimeError	If the active project is not set.

Source code in src/zenml/config/global_config.py

def get_active_project_id(self) -> UUID:
    """Get the ID of the active project.

    Returns:
        The ID of the active project.

    Raises:
        RuntimeError: If the active project is not set.
    """
    if self.active_project_id is None:
        _ = self.zen_store
        if self.active_project_id is None:
            raise RuntimeError(
                "No project is currently set as active. Please set the "
                "active project using the `zenml project set <NAME>` CLI "
                "command."
            )

    return self.active_project_id

get_active_stack_id() -> UUID

Get the ID of the active stack.

If the active stack doesn't exist yet, the ZenStore is reinitialized.

Returns:

Type	Description
UUID	The active stack ID.

Source code in src/zenml/config/global_config.py

def get_active_stack_id(self) -> UUID:
    """Get the ID of the active stack.

    If the active stack doesn't exist yet, the ZenStore is reinitialized.

    Returns:
        The active stack ID.
    """
    if self.active_stack_id is None:
        _ = self.zen_store
        assert self.active_stack_id is not None

    return self.active_stack_id

get_config_environment_vars() -> Dict[str, str]

Convert the global configuration to environment variables.

Returns:

Type	Description
Dict[str, str]	Environment variables dictionary.

Source code in src/zenml/config/global_config.py

def get_config_environment_vars(self) -> Dict[str, str]:
    """Convert the global configuration to environment variables.

    Returns:
        Environment variables dictionary.
    """
    environment_vars = {}

    for key in type(self).model_fields.keys():
        if key == "store":
            # The store configuration uses its own environment variable
            # naming scheme
            continue

        value = getattr(self, key)
        if value is not None:
            environment_vars[CONFIG_ENV_VAR_PREFIX + key.upper()] = str(
                value
            )

    store_dict = self.store_configuration.model_dump(exclude_none=True)

    # The secrets store and backup secrets store configurations use their
    # own environment variables naming scheme
    secrets_store_dict = store_dict.pop("secrets_store", None) or {}
    backup_secrets_store_dict = (
        store_dict.pop("backup_secrets_store", None) or {}
    )

    for key, value in store_dict.items():
        if key in ["username", "password"]:
            # Never include the username and password in the env vars. Use
            # the API token instead.
            continue

        environment_vars[ENV_ZENML_STORE_PREFIX + key.upper()] = str(value)

    for key, value in secrets_store_dict.items():
        environment_vars[ENV_ZENML_SECRETS_STORE_PREFIX + key.upper()] = (
            str(value)
        )

    for key, value in backup_secrets_store_dict.items():
        environment_vars[
            ENV_ZENML_BACKUP_SECRETS_STORE_PREFIX + key.upper()
        ] = str(value)

    return environment_vars

get_default_store() -> StoreConfiguration

Get the default SQLite store configuration.

Returns:

Type	Description
StoreConfiguration	The default SQLite store configuration.

Source code in src/zenml/config/global_config.py

def get_default_store(self) -> StoreConfiguration:
    """Get the default SQLite store configuration.

    Returns:
        The default SQLite store configuration.
    """
    from zenml.zen_stores.base_zen_store import BaseZenStore

    return BaseZenStore.get_default_store_config(
        path=os.path.join(
            self.local_stores_path,
            DEFAULT_STORE_DIRECTORY_NAME,
        )
    )

get_instance() -> Optional[GlobalConfiguration] classmethod

Return the GlobalConfiguration singleton instance.

Returns:

Type	Description
Optional[GlobalConfiguration]	The GlobalConfiguration singleton instance or None, if the
Optional[GlobalConfiguration]	GlobalConfiguration hasn't been initialized yet.

Source code in src/zenml/config/global_config.py

@classmethod
def get_instance(cls) -> Optional["GlobalConfiguration"]:
    """Return the GlobalConfiguration singleton instance.

    Returns:
        The GlobalConfiguration singleton instance or None, if the
        GlobalConfiguration hasn't been initialized yet.
    """
    return cls._global_config

set_active_project(project: ProjectResponse) -> ProjectResponse

Set the project for the local client.

Parameters:

Name	Type	Description	Default
project	ProjectResponse	The project to set active.	required

Returns:

Type	Description
ProjectResponse	The project that was set active.

Source code in src/zenml/config/global_config.py

def set_active_project(
    self, project: "ProjectResponse"
) -> "ProjectResponse":
    """Set the project for the local client.

    Args:
        project: The project to set active.

    Returns:
        The project that was set active.
    """
    self.active_project_id = project.id
    self._active_project = project
    # Sanitize the global configuration to reflect the new project
    self._sanitize_config()
    return project

set_active_stack(stack: StackResponse) -> None

Set the active stack for the local client.

Parameters:

Name	Type	Description	Default
stack	StackResponse	The model of the stack to set active.	required

Source code in src/zenml/config/global_config.py

def set_active_stack(self, stack: "StackResponse") -> None:
    """Set the active stack for the local client.

    Args:
        stack: The model of the stack to set active.
    """
    self.active_stack_id = stack.id
    self._active_stack = stack

set_default_store() -> None

Initializes and sets the default store configuration.

Call this method to initialize or revert the store configuration to the default store.

Source code in src/zenml/config/global_config.py

def set_default_store(self) -> None:
    """Initializes and sets the default store configuration.

    Call this method to initialize or revert the store configuration to the
    default store.
    """
    # Apply the environment variables to the default store configuration
    default_store_cfg = self._get_store_configuration(
        baseline=self.get_default_store()
    )
    self._configure_store(default_store_cfg)
    logger.debug("Using the default store for the global config.")

set_store(config: StoreConfiguration, skip_default_registrations: bool = False, **kwargs: Any) -> None

Update the active store configuration.

Call this method to validate and update the active store configuration.

Parameters:

Name	Type	Description	Default
config	StoreConfiguration	The new store configuration to use.	required
skip_default_registrations	bool	If True, the creation of the default stack and user in the store will be skipped.	False
**kwargs	Any	Additional keyword arguments to pass to the store constructor.	{}

Source code in src/zenml/config/global_config.py

def set_store(
    self,
    config: StoreConfiguration,
    skip_default_registrations: bool = False,
    **kwargs: Any,
) -> None:
    """Update the active store configuration.

    Call this method to validate and update the active store configuration.

    Args:
        config: The new store configuration to use.
        skip_default_registrations: If `True`, the creation of the default
            stack and user in the store will be skipped.
        **kwargs: Additional keyword arguments to pass to the store
            constructor.
    """
    # Apply the environment variables to the custom store configuration
    config = self._get_store_configuration(baseline=config)
    self._configure_store(config, skip_default_registrations, **kwargs)
    logger.info("Updated the global store configuration.")

uses_default_store() -> bool

Check if the global configuration uses the default store.

Returns:

Type	Description
bool	True if the global configuration uses the default store.

Source code in src/zenml/config/global_config.py

def uses_default_store(self) -> bool:
    """Check if the global configuration uses the default store.

    Returns:
        `True` if the global configuration uses the default store.
    """
    return self.store_configuration.url == self.get_default_store().url

Functions

Modules

pipeline_configurations

Pipeline configuration classes.

Classes

PipelineConfiguration

Bases: PipelineConfigurationUpdate

Pipeline configuration class.

Attributes

docker_settings: DockerSettings property

Docker settings of this pipeline.

Returns:

Type	Description
DockerSettings	The Docker settings of this pipeline.

Functions

ensure_pipeline_name_allowed(name: str) -> str classmethod

Ensures the pipeline name is allowed.

Parameters:

Name	Type	Description	Default
name	str	Name of the pipeline.	required

Returns:

Type	Description
str	The validated name of the pipeline.

Raises:

Type	Description
ValueError	If the name is not allowed.

Source code in src/zenml/config/pipeline_configurations.py

@field_validator("name")
@classmethod
def ensure_pipeline_name_allowed(cls, name: str) -> str:
    """Ensures the pipeline name is allowed.

    Args:
        name: Name of the pipeline.

    Returns:
        The validated name of the pipeline.

    Raises:
        ValueError: If the name is not allowed.
    """
    if name in DISALLOWED_PIPELINE_NAMES:
        raise ValueError(
            f"Pipeline name '{name}' is not allowed since '{name}' is a "
            "reserved key word. Please choose another name."
        )
    return name

PipelineConfigurationUpdate

Bases: StrictBaseModel

Class for pipeline configuration updates.

Functions

pipeline_run_configuration

Pipeline run configuration class.

Classes

PipelineRunConfiguration

Bases: StrictBaseModel, YAMLSerializationMixin

Class for pipeline run configurations.

Modules

pipeline_spec

Pipeline configuration classes.

Classes

PipelineSpec

Bases: StrictBaseModel

Specification of a pipeline.

Attributes

json_with_string_sources: str property

JSON representation with sources replaced by their import path.

Returns:

Type	Description
str	The JSON representation.

resource_settings

Resource settings class used to specify resources for a step.

Classes

ByteUnit

Bases: Enum

Enum for byte units.

Attributes

byte_value: int property

Returns the amount of bytes that this unit represents.

Returns:

Type	Description
int	The byte value of this unit.

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

Bases: BaseSettings

Hardware resource settings.

Attributes:

Name	Type	Description
cpu_count	Optional[PositiveFloat]	The amount of CPU cores that should be configured.
gpu_count	Optional[NonNegativeInt]	The amount of GPUs that should be configured.
memory	Optional[str]	The amount of memory that should be configured.

Source code in src/zenml/config/secret_reference_mixin.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references are only passed for valid fields.

    This method ensures that secret references are not passed for fields
    that explicitly prevent them or require pydantic validation.

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

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            or an attribute which explicitly disallows secret references
            is 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}`. 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 values with secrets "
                    "here: https://docs.zenml.io/getting-started/deploying-zenml/secret-management"
                )
            continue

        if secret_utils.is_clear_text_field(field):
            raise ValueError(
                f"Passing the `{key}` attribute as a secret reference is "
                "not allowed."
            )

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

    super().__init__(**kwargs)

Attributes

empty: bool property

Returns if this object is "empty" (=no values configured) or not.

Returns:

Type	Description
bool	True if no values were configured, False otherwise.

Functions

get_memory(unit: Union[str, ByteUnit] = ByteUnit.GB) -> Optional[float]

Gets the memory configuration in a specific unit.

Parameters:

Name	Type	Description	Default
unit	Union[str, ByteUnit]	The unit to which the memory should be converted.	GB

Raises:

Type	Description
ValueError	If the memory string is invalid.

Returns:

Type	Description
Optional[float]	The memory configuration converted to the requested unit, or None
Optional[float]	if no memory was configured.

Source code in src/zenml/config/resource_settings.py

def get_memory(
    self, unit: Union[str, ByteUnit] = ByteUnit.GB
) -> Optional[float]:
    """Gets the memory configuration in a specific unit.

    Args:
        unit: The unit to which the memory should be converted.

    Raises:
        ValueError: If the memory string is invalid.

    Returns:
        The memory configuration converted to the requested unit, or None
        if no memory was configured.
    """
    if not self.memory:
        return None

    if isinstance(unit, str):
        unit = ByteUnit(unit)

    memory = self.memory
    for memory_unit in ByteUnit:
        if memory.endswith(memory_unit.value):
            memory_value = int(memory[: -len(memory_unit.value)])
            return memory_value * memory_unit.byte_value / unit.byte_value
    else:
        # Should never happen due to the regex validation
        raise ValueError(f"Unable to parse memory unit from '{memory}'.")

retry_config

Retry configuration for a step.

Classes

StepRetryConfig

Bases: StrictBaseModel

Retry configuration for a step.

Delay is an integer (specified in seconds).

schedule

Class for defining a pipeline schedule.

Classes

Schedule

Bases: BaseModel

Class for defining a pipeline schedule.

Attributes:

Name	Type	Description
name	Optional[str]	Optional name to give to the schedule. If not set, a default name will be generated based on the pipeline name and the current date and time.
cron_expression	Optional[str]	Cron expression for the pipeline schedule. If a value for this is set it takes precedence over the start time + interval.
start_time	Optional[datetime]	When the schedule should start. If this is a datetime object without any timezone, it is treated as a datetime in the local timezone.
end_time	Optional[datetime]	When the schedule should end. If this is a datetime object without any timezone, it is treated as a datetime in the local timezone.
interval_second	Optional[timedelta]	datetime timedelta indicating the seconds between two recurring runs for a periodic schedule.
catchup	bool	Whether the recurring run should catch up if behind schedule. For example, if the recurring run is paused for a while and re-enabled afterward. If catchup=True, the scheduler will catch up on (backfill) each missed interval. Otherwise, it only schedules the latest interval if more than one interval is ready to be scheduled. Usually, if your pipeline handles backfill internally, you should turn catchup off to avoid duplicate backfill.
run_once_start_time	Optional[datetime]	When to run the pipeline once. If this is a datetime object without any timezone, it is treated as a datetime in the local timezone.

Functions

secret_reference_mixin

Secret reference mixin implementation.

Classes

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

Bases: BaseModel

Mixin class for secret references in pydantic model attributes.

Ensures that secret references are only passed for valid fields.

This method ensures that secret references are not passed for fields that explicitly prevent them or require pydantic validation.

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

Raises:

Type	Description
ValueError	If an attribute that requires custom pydantic validation or an attribute which explicitly disallows secret references is passed as a secret reference.

Source code in src/zenml/config/secret_reference_mixin.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references are only passed for valid fields.

    This method ensures that secret references are not passed for fields
    that explicitly prevent them or require pydantic validation.

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

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            or an attribute which explicitly disallows secret references
            is 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}`. 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 values with secrets "
                    "here: https://docs.zenml.io/getting-started/deploying-zenml/secret-management"
                )
            continue

        if secret_utils.is_clear_text_field(field):
            raise ValueError(
                f"Passing the `{key}` attribute as a secret reference is "
                "not allowed."
            )

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

    super().__init__(**kwargs)

Attributes

required_secrets: Set[secret_utils.SecretReference] property

All required secrets for this object.

Returns:

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

Functions

Functions

Modules

secrets_store_config

Functionality to support ZenML secrets store configurations.

Classes

SecretsStoreConfiguration

Bases: BaseModel

Generic secrets store configuration.

The store configurations of concrete secrets store implementations must inherit from this class and validate any extra attributes that are configured in addition to those defined in this class.

Attributes:

Name	Type	Description
type	SecretsStoreType	The type of store backend.
class_path	Optional[str]	The Python class path of the store backend. Should point to a subclass of BaseSecretsStore. This is optional and only required if the store backend is not one of the built-in implementations.

Functions

validate_custom() -> SecretsStoreConfiguration

Validate that class_path is set for custom secrets stores.

Returns:

Type	Description
SecretsStoreConfiguration	Validated settings.

Raises:

Type	Description
ValueError	If class_path is not set when using a custom secrets store.

Source code in src/zenml/config/secrets_store_config.py

@model_validator(mode="after")
def validate_custom(self) -> "SecretsStoreConfiguration":
    """Validate that class_path is set for custom secrets stores.

    Returns:
        Validated settings.

    Raises:
        ValueError: If class_path is not set when using a custom secrets
            store.
    """
    if not self.type:
        return self
    if self.type == SecretsStoreType.CUSTOM:
        if self.class_path is None:
            raise ValueError(
                "A class_path must be set when using a custom secrets "
                "store implementation."
            )
    elif self.class_path is not None:
        raise ValueError(
            f"The class_path attribute is not supported for the "
            f"{self.type} secrets store type."
        )

    return self

Functions

server_config

Functionality to support ZenML Server Configuration.

Classes

ServerConfiguration

Bases: BaseModel

ZenML Server configuration attributes.

All these attributes can be set through the environment with the ZENML_SERVER_-Prefix. The value of the ZENML_SERVER_DEPLOYMENT_TYPE environment variable will be extracted to deployment_type.

Attributes:

Name	Type	Description
deployment_type	ServerDeploymentType	The type of ZenML server deployment that is running.
server_url	Optional[str]	The URL where the ZenML server API is reachable. Must be configured for features that involve triggering workloads from the ZenML dashboard (e.g., running pipelines). If not specified, the clients will use the same URL used to connect them to the ZenML server.
dashboard_url	Optional[str]	The URL where the ZenML dashboard is reachable. If not specified, the server_url value is used. This should be configured if the dashboard is served from a different URL than the ZenML server.
root_url_path	str	The root URL path for the ZenML API and dashboard.
auth_scheme	AuthScheme	The authentication scheme used by the ZenML server.
jwt_token_algorithm	str	The algorithm used to sign and verify JWT tokens.
jwt_token_issuer	Optional[str]	The issuer of the JWT tokens. If not specified, the issuer is set to the ZenML Server ID.
jwt_token_audience	Optional[str]	The audience of the JWT tokens. If not specified, the audience is set to the ZenML Server ID.
jwt_token_leeway_seconds	int	The leeway in seconds allowed when verifying the expiration time of JWT tokens.
jwt_token_expire_minutes	Optional[int]	The expiration time of JWT tokens in minutes. If not specified, generated JWT tokens will not be set to expire.
jwt_secret_key	str	The secret key used to sign and verify JWT tokens. If not specified, a random secret key is generated.
auth_cookie_name	Optional[str]	The name of the http-only cookie used to store the JWT token. If not specified, the cookie name is set to a value computed from the ZenML server ID.
auth_cookie_domain	Optional[str]	The domain of the http-only cookie used to store the JWT token. If not specified, the cookie will be valid for the domain where the ZenML server is running.
cors_allow_origins	Optional[List[str]]	The origins allowed to make cross-origin requests to the ZenML server. If not specified, all origins are allowed.
max_failed_device_auth_attempts	int	The maximum number of failed OAuth 2.0 device authentication attempts before the device is locked.
device_auth_timeout	int	The timeout in seconds after which a pending OAuth 2.0 device authorization request expires.
device_auth_polling_interval	int	The polling interval in seconds used to poll the OAuth 2.0 device authorization endpoint.
device_expiration_minutes	Optional[int]	The time in minutes that an OAuth 2.0 device is allowed to be used to authenticate with the ZenML server. If not set or if jwt_token_expire_minutes is not set, the devices are allowed to be used indefinitely. This controls the expiration time of the JWT tokens issued to clients after they have authenticated with the ZenML server using an OAuth 2.0 device.
trusted_device_expiration_minutes	Optional[int]	The time in minutes that a trusted OAuth 2.0 device is allowed to be used to authenticate with the ZenML server. If not set or if jwt_token_expire_minutes is not set, the devices are allowed to be used indefinitely. This controls the expiration time of the JWT tokens issued to clients after they have authenticated with the ZenML server using an OAuth 2.0 device that has been marked as trusted.
generic_api_token_lifetime	PositiveInt	The lifetime in seconds that generic short-lived API tokens issued for automation purposes are valid.
external_login_url	Optional[str]	The login URL of an external authenticator service to use with the EXTERNAL authentication scheme.
external_user_info_url	Optional[str]	The user info URL of an external authenticator service to use with the EXTERNAL authentication scheme.
external_server_id	Optional[UUID]	The ID of the ZenML server to use with the EXTERNAL authentication scheme. If not specified, the regular ZenML server ID is used.
metadata	Dict[str, str]	Additional metadata to be associated with the ZenML server.
rbac_implementation_source	Optional[str]	Source pointing to a class implementing the RBAC interface defined by zenml.zen_server.rbac_interface.RBACInterface. If not specified, RBAC will not be enabled for this server.
feature_gate_implementation_source	Optional[str]	Source pointing to a class implementing the feature gate interface defined by zenml.zen_server.feature_gate.feature_gate_interface.FeatureGateInterface. If not specified, feature usage will not be gated/tracked for this server.
workload_manager_implementation_source	Optional[str]	Source pointing to a class implementing the workload management interface.
pipeline_run_auth_window	int	The default time window in minutes for which a pipeline run action is allowed to authenticate with the ZenML server.
login_rate_limit_minute	int	The number of login attempts allowed per minute.
login_rate_limit_day	int	The number of login attempts allowed per day.
secure_headers_server	Union[bool, str]	Custom value to be set in the Server HTTP header to identify the server. If not specified, or if set to one of the reserved values enabled, yes, true, on, the Server header will be set to the default value (ZenML server ID). If set to one of the reserved values disabled, no, none, false, off or to an empty string, the Server header will not be included in responses.
secure_headers_hsts	Union[bool, str]	The server header value to be set in the HTTP header Strict-Transport-Security. If not specified, or if set to one of the reserved values enabled, yes, true, on, the Strict-Transport-Security header will be set to the default value (max-age=63072000; includeSubdomains). If set to one of the reserved values disabled, no, none, false, off or to an empty string, the Strict-Transport-Security header will not be included in responses.
secure_headers_xfo	Union[bool, str]	The server header value to be set in the HTTP header X-Frame-Options. If not specified, or if set to one of the reserved values enabled, yes, true, on, the X-Frame-Options header will be set to the default value (SAMEORIGIN). If set to one of the reserved values disabled, no, none, false, off or to an empty string, the X-Frame-Options header will not be included in responses.
secure_headers_xxp	Union[bool, str]	The server header value to be set in the HTTP header X-XSS-Protection. If not specified, or if set to one of the reserved values enabled, yes, true, on, the X-XSS-Protection header will be set to the default value (0). If set to one of the reserved values disabled, no, none, false, off or to an empty string, the X-XSS-Protection header will not be included in responses. NOTE: this header is deprecated and should always be set to 0. The Content-Security-Policy header should be used instead.
secure_headers_content	Union[bool, str]	The server header value to be set in the HTTP header X-Content-Type-Options. If not specified, or if set to one of the reserved values enabled, yes, true, on, the X-Content-Type-Options header will be set to the default value (nosniff). If set to one of the reserved values disabled, no, none, false, off or to an empty string, the X-Content-Type-Options header will not be included in responses.
secure_headers_csp	Union[bool, str]	The server header value to be set in the HTTP header Content-Security-Policy. If not specified, or if set to one of the reserved values enabled, yes, true, on, the Content-Security-Policy header will be set to a default value that is compatible with the ZenML dashboard. If set to one of the reserved values disabled, no, none, false, off or to an empty string, the Content-Security-Policy header will not be included in responses.
secure_headers_referrer	Union[bool, str]	The server header value to be set in the HTTP header Referrer-Policy. If not specified, or if set to one of the reserved values enabled, yes, true, on, the Referrer-Policy header will be set to the default value (no-referrer-when-downgrade). If set to one of the reserved values disabled, no, none, false, off or to an empty string, the Referrer-Policy header will not be included in responses.
secure_headers_cache	Union[bool, str]	The server header value to be set in the HTTP header Cache-Control. If not specified, or if set to one of the reserved values enabled, yes, true, on, the Cache-Control header will be set to the default value (no-store, no-cache, must-revalidate). If set to one of the reserved values disabled, no, none, false, off or to an empty string, the Cache-Control header will not be included in responses.
secure_headers_permissions	Union[bool, str]	The server header value to be set in the HTTP header Permissions-Policy. If not specified, or if set to one of the reserved values enabled, yes, true, on, the Permissions-Policy header will be set to the default value (accelerometer=(), camera=(), geolocation=(), gyroscope=(), magnetometer=(), microphone=(), payment=(), usb=()). If set to one of the reserved values disabled, no, none, false, off or to an empty string, the Permissions-Policy header will not be included in responses.
server_name	str	The name of the ZenML server. Used only during initial deployment. Can be changed later as a part of the server settings.
display_announcements	bool	Whether to display announcements about ZenML in the dashboard. Used only during initial deployment. Can be changed later as a part of the server settings.
display_updates	bool	Whether to display notifications about ZenML updates in the dashboard. Used only during initial deployment. Can be changed later as a part of the server settings.
auto_activate	bool	Whether to automatically activate the server and create a default admin user account with an empty password during the initial deployment.
max_request_body_size_in_bytes	int	The maximum size of the request body in bytes. If not specified, the default value of 256 Kb will be used.
memcache_max_capacity	int	The maximum number of entries that the memory cache can hold. If not specified, the default value of 1000 will be used.
memcache_default_expiry	int	The default expiry time in seconds for cache entries. If not specified, the default value of 30 seconds will be used.
file_download_size_limit	int	The maximum size of the file download in bytes. If not specified, the default value of 2GB will be used.

Attributes

deployment_id: UUID property

Get the ZenML server deployment ID.

Returns:

Type	Description
UUID	The ZenML server deployment ID.

feature_gate_enabled: bool property

Whether feature gating is enabled on the server or not.

Returns:

Type	Description
bool	Whether feature gating is enabled on the server or not.

rbac_enabled: bool property

Whether RBAC is enabled on the server or not.

Returns:

Type	Description
bool	Whether RBAC is enabled on the server or not.

workload_manager_enabled: bool property

Whether workload management is enabled on the server or not.

Returns:

Type	Description
bool	Whether workload management is enabled on the server or not.

Functions

get_auth_cookie_name() -> str

Get the authentication cookie name.

If not configured, the cookie name is set to a value computed from the ZenML server ID.

Returns:

Type	Description
str	The authentication cookie name.

Source code in src/zenml/config/server_config.py

def get_auth_cookie_name(self) -> str:
    """Get the authentication cookie name.

    If not configured, the cookie name is set to a value computed from the
    ZenML server ID.

    Returns:
        The authentication cookie name.
    """
    if self.auth_cookie_name:
        return self.auth_cookie_name

    self.auth_cookie_name = f"zenml-server-{self.deployment_id}"

    return self.auth_cookie_name

get_external_server_id() -> UUID

Get the external server ID.

If not configured, the regular ZenML server ID is used.

Returns:

Type	Description
UUID	The external server ID.

Source code in src/zenml/config/server_config.py

def get_external_server_id(self) -> UUID:
    """Get the external server ID.

    If not configured, the regular ZenML server ID is used.

    Returns:
        The external server ID.
    """
    if self.external_server_id:
        return self.external_server_id

    self.external_server_id = self.deployment_id

    return self.external_server_id

get_jwt_token_audience() -> str

Get the JWT token audience.

If not configured, the audience is set to the ZenML Server ID.

Returns:

Type	Description
str	The JWT token audience.

Source code in src/zenml/config/server_config.py

def get_jwt_token_audience(self) -> str:
    """Get the JWT token audience.

    If not configured, the audience is set to the ZenML Server ID.

    Returns:
        The JWT token audience.
    """
    if self.jwt_token_audience:
        return self.jwt_token_audience

    self.jwt_token_audience = str(self.deployment_id)

    return self.jwt_token_audience

get_jwt_token_issuer() -> str

Get the JWT token issuer.

If not configured, the issuer is set to the ZenML Server ID.

Returns:

Type	Description
str	The JWT token issuer.

Source code in src/zenml/config/server_config.py

def get_jwt_token_issuer(self) -> str:
    """Get the JWT token issuer.

    If not configured, the issuer is set to the ZenML Server ID.

    Returns:
        The JWT token issuer.
    """
    if self.jwt_token_issuer:
        return self.jwt_token_issuer

    self.jwt_token_issuer = str(self.deployment_id)

    return self.jwt_token_issuer

get_server_config() -> ServerConfiguration classmethod

Get the server configuration.

Returns:

Type	Description
ServerConfiguration	The server configuration.

Source code in src/zenml/config/server_config.py

@classmethod
def get_server_config(cls) -> "ServerConfiguration":
    """Get the server configuration.

    Returns:
        The server configuration.
    """
    env_server_config: Dict[str, Any] = {}
    for k, v in os.environ.items():
        if v == "":
            continue
        if k.startswith(ENV_ZENML_SERVER_PRO_PREFIX):
            # Skip Pro configuration
            continue
        if k.startswith(ENV_ZENML_SERVER_PREFIX):
            env_server_config[
                k[len(ENV_ZENML_SERVER_PREFIX) :].lower()
            ] = v

    server_config = ServerConfiguration(**env_server_config)

    if server_config.deployment_type == ServerDeploymentType.CLOUD:
        # If the zenml server is a Pro server, we will apply the Pro
        # configuration overrides to the server config automatically.
        # TODO: these should be retrieved dynamically from the ZenML Pro
        # API.
        server_pro_config = ServerProConfiguration.get_server_config()
        server_config.auth_scheme = AuthScheme.EXTERNAL
        server_config.external_login_url = (
            f"{server_pro_config.dashboard_url}/api/auth/login"
        )
        server_config.external_user_info_url = (
            f"{server_pro_config.api_url}/users/authorize_server"
        )
        server_config.external_server_id = server_pro_config.workspace_id
        server_config.rbac_implementation_source = (
            "zenml.zen_server.rbac.zenml_cloud_rbac.ZenMLCloudRBAC"
        )
        server_config.feature_gate_implementation_source = "zenml.zen_server.feature_gate.zenml_cloud_feature_gate.ZenMLCloudFeatureGateInterface"
        server_config.reportable_resources = DEFAULT_REPORTABLE_RESOURCES
        server_config.dashboard_url = f"{server_pro_config.dashboard_url}/workspaces/{server_pro_config.workspace_name}"
        server_config.metadata.update(
            dict(
                account_id=str(server_pro_config.organization_id),
                organization_id=str(server_pro_config.organization_id),
                workspace_id=str(server_pro_config.workspace_id),
            )
        )
        if server_pro_config.workspace_name:
            server_config.metadata.update(
                dict(workspace_name=server_pro_config.workspace_name)
            )

        extra_cors_allow_origins = [
            server_pro_config.dashboard_url,
            server_pro_config.api_url,
        ]
        if server_config.server_url:
            extra_cors_allow_origins.append(server_config.server_url)
        if (
            not server_config.cors_allow_origins
            or server_config.cors_allow_origins == ["*"]
        ):
            server_config.cors_allow_origins = extra_cors_allow_origins
        else:
            server_config.cors_allow_origins += extra_cors_allow_origins
            if "*" in server_config.cors_allow_origins:
                # Remove the wildcard from the list
                server_config.cors_allow_origins.remove("*")

            # Remove duplicates
            server_config.cors_allow_origins = list(
                set(server_config.cors_allow_origins)
            )

        if server_config.jwt_token_expire_minutes is None:
            server_config.jwt_token_expire_minutes = 60

    return server_config

ServerProConfiguration

Bases: BaseModel

ZenML Server Pro configuration attributes.

All these attributes can be set through the environment with the ZENML_SERVER_PRO_-Prefix. E.g. the value of the ZENML_SERVER_PRO_API_URL environment variable will be extracted to api_url.

Attributes:

Name	Type	Description
api_url	str	The ZenML Pro API URL.
dashboard_url	str	The ZenML Pro dashboard URL.
oauth2_client_secret	str	The ZenML Pro OAuth2 client secret used to authenticate the ZenML server with the ZenML Pro API.
oauth2_audience	str	The OAuth2 audience.
organization_id	UUID	The ZenML Pro organization ID.
organization_name	Optional[str]	The ZenML Pro organization name.
workspace_id	UUID	The ZenML Pro workspace ID.
workspace_name	Optional[str]	The ZenML Pro workspace name.

Functions

get_server_config() -> ServerProConfiguration classmethod

Get the server Pro configuration.

Returns:

Type	Description
ServerProConfiguration	The server Pro configuration.

Source code in src/zenml/config/server_config.py

@classmethod
def get_server_config(cls) -> "ServerProConfiguration":
    """Get the server Pro configuration.

    Returns:
        The server Pro configuration.
    """
    env_server_config: Dict[str, Any] = {}
    for k, v in os.environ.items():
        if v == "":
            continue
        if k.startswith(ENV_ZENML_SERVER_PRO_PREFIX):
            env_server_config[
                k[len(ENV_ZENML_SERVER_PRO_PREFIX) :].lower()
            ] = v

    return ServerProConfiguration(**env_server_config)

Functions

generate_jwt_secret_key() -> str

Generate a random JWT secret key.

This key is used to sign and verify generated JWT tokens.

Returns:

Type	Description
str	A random JWT secret key.

Source code in src/zenml/config/server_config.py

def generate_jwt_secret_key() -> str:
    """Generate a random JWT secret key.

    This key is used to sign and verify generated JWT tokens.

    Returns:
        A random JWT secret key.
    """
    return token_hex(32)

settings_resolver

Class for resolving settings.

Classes

SettingsResolver(key: str, settings: BaseSettings)

Class for resolving settings.

This class converts a BaseSettings instance to the correct subclass depending on the key for which these settings were specified.

Checks if the settings key is valid.

Parameters:

Name	Type	Description	Default
key	str	Settings key.	required
settings	BaseSettings	The settings.	required

Raises:

Type	Description
ValueError	If the settings key is invalid.

Source code in src/zenml/config/settings_resolver.py

def __init__(self, key: str, settings: "BaseSettings"):
    """Checks if the settings key is valid.

    Args:
        key: Settings key.
        settings: The settings.

    Raises:
        ValueError: If the settings key is invalid.
    """
    if not settings_utils.is_valid_setting_key(key):
        raise ValueError(
            f"Invalid setting key `{key}`. Setting keys can either refer "
            "to general settings (available keys: "
            f"{set(settings_utils.get_general_settings())}) or stack "
            "component specific settings. Stack component specific keys "
            "are of the format "
            "`<STACK_COMPONENT_TYPE>.<STACK_COMPONENT_FLAVOR>`."
        )

    self._key = key
    self._settings = settings

Functions

resolve(stack: Stack) -> BaseSettings

Resolves settings for the given stack.

Parameters:

Name	Type	Description	Default
stack	Stack	The stack for which to resolve the settings.	required

Returns:

Type	Description
BaseSettings	The resolved settings.

Source code in src/zenml/config/settings_resolver.py

def resolve(self, stack: "Stack") -> "BaseSettings":
    """Resolves settings for the given stack.

    Args:
        stack: The stack for which to resolve the settings.

    Returns:
        The resolved settings.
    """
    if settings_utils.is_general_setting_key(self._key):
        target_class = self._resolve_general_settings_class()
    else:
        target_class = self._resolve_stack_component_setting_class(
            stack=stack
        )

    return self._convert_settings(target_class=target_class)

Modules

source

Source classes.

Classes

CodeRepositorySource

Bases: Source

Source representing an object from a code repository.

Attributes:

Name	Type	Description
repository_id	UUID	The code repository ID.
commit	str	The commit.
subdirectory	str	The subdirectory of the source root inside the code repository.

DistributionPackageSource

Bases: Source

Source representing an object from a distribution package.

Attributes:

Name	Type	Description
package_name	str	Name of the package.
version	Optional[str]	The package version.

NotebookSource

Bases: Source

Source representing an object defined in a notebook.

Attributes:

Name	Type	Description
replacement_module	Optional[str]	Name of the module from which this source should be loaded in case the code is not running in a notebook.
artifact_store_id	Optional[UUID]	ID of the artifact store in which the replacement module code is stored.

Source

Bases: BaseModel

Source specification.

A source specifies a module name as well as an optional attribute of that module. These values can be used to import the module and get the value of the attribute inside the module.

Example

The source Source(module="zenml.config.source", attribute="Source") references the class that this docstring is describing. This class is defined in the zenml.config.source module and the name of the attribute is the class name Source.

Attributes:

Name	Type	Description
module	str	The module name.
attribute	Optional[str]	Optional name of the attribute inside the module.
type	SourceType	The type of the source.

Attributes

import_path: str property

The import path of the source.

Returns:

Type	Description
str	The import path of the source.

is_internal: bool property

If the source is internal (=from the zenml package).

Returns:

Type	Description
bool	True if the source is internal, False otherwise

is_module_source: bool property

If the source is a module source.

Returns:

Type	Description
bool	If the source is a module source.

Functions

from_import_path(import_path: str, is_module_path: bool = False) -> Source classmethod

Creates a source from an import path.

Parameters:

Name	Type	Description	Default
import_path	str	The import path.	required
is_module_path	bool	If the import path points to a module or not.	False

Raises:

Type	Description
ValueError	If the import path is empty.

Returns:

Type	Description
Source	The source.

Source code in src/zenml/config/source.py

@classmethod
def from_import_path(
    cls, import_path: str, is_module_path: bool = False
) -> "Source":
    """Creates a source from an import path.

    Args:
        import_path: The import path.
        is_module_path: If the import path points to a module or not.

    Raises:
        ValueError: If the import path is empty.

    Returns:
        The source.
    """
    if not import_path:
        raise ValueError(
            "Invalid empty import path. The import path needs to refer "
            "to a Python module and an optional attribute of that module."
        )

    # Remove internal version pins for backwards compatibility
    if "@" in import_path:
        import_path = import_path.split("@", 1)[0]

    if is_module_path or "." not in import_path:
        module = import_path
        attribute = None
    else:
        module, attribute = import_path.rsplit(".", maxsplit=1)

    return Source(
        module=module, attribute=attribute, type=SourceType.UNKNOWN
    )

model_dump(**kwargs: Any) -> Dict[str, Any]

Dump the source as a dictionary.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Additional keyword arguments.	{}

Returns:

Type	Description
Dict[str, Any]	The source as a dictionary.

Source code in src/zenml/config/source.py

def model_dump(self, **kwargs: Any) -> Dict[str, Any]:
    """Dump the source as a dictionary.

    Args:
        **kwargs: Additional keyword arguments.

    Returns:
        The source as a dictionary.
    """
    return super().model_dump(serialize_as_any=True, **kwargs)

model_dump_json(**kwargs: Any) -> str

Dump the source as a JSON string.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Additional keyword arguments.	{}

Returns:

Type	Description
str	The source as a JSON string.

Source code in src/zenml/config/source.py

def model_dump_json(self, **kwargs: Any) -> str:
    """Dump the source as a JSON string.

    Args:
        **kwargs: Additional keyword arguments.

    Returns:
        The source as a JSON string.
    """
    return super().model_dump_json(serialize_as_any=True, **kwargs)

SourceType

Bases: Enum

Enum representing different types of sources.

Functions

convert_source(source: Any) -> Any

Converts an old source string to a source object.

Parameters:

Name	Type	Description	Default
source	Any	Source string or object.	required

Returns:

Type	Description
Any	The converted source.

Source code in src/zenml/config/source.py

def convert_source(source: Any) -> Any:
    """Converts an old source string to a source object.

    Args:
        source: Source string or object.

    Returns:
        The converted source.
    """
    if isinstance(source, str):
        source = Source.from_import_path(source)

    return source

step_configurations

Pipeline configuration classes.

Classes

ArtifactConfiguration

Bases: PartialArtifactConfiguration

Class representing a complete input/output artifact configuration.

InputSpec

Bases: StrictBaseModel

Step input specification.

PartialArtifactConfiguration

Bases: StrictBaseModel

Class representing a partial input/output artifact configuration.

PartialStepConfiguration

Bases: StepConfigurationUpdate

Class representing a partial step configuration.

Step

Bases: StrictBaseModel

Class representing a ZenML step.

StepConfiguration

Bases: PartialStepConfiguration

Step configuration class.

Attributes

docker_settings: DockerSettings property

Docker settings of this step configuration.

Returns:

Type	Description
DockerSettings	The Docker settings of this step configuration.

resource_settings: ResourceSettings property

Resource settings of this step configuration.

Returns:

Type	Description
ResourceSettings	The resource settings of this step configuration.

StepConfigurationUpdate

Bases: StrictBaseModel

Class for step configuration updates.

StepSpec

Bases: StrictBaseModel

Specification of a pipeline.

Functions

Modules

step_run_info

Step run info.

Classes

StepRunInfo

Bases: StrictBaseModel

All information necessary to run a step.

Functions

get_image(key: str) -> str

Gets the Docker image for the given key.

Parameters:

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

Raises:

Type	Description
RuntimeError	If the run does not have an associated build.

Returns:

Type	Description
str	The image name or digest.

Source code in src/zenml/config/step_run_info.py

def get_image(self, key: str) -> str:
    """Gets the Docker image for the given key.

    Args:
        key: The key for which to get the image.

    Raises:
        RuntimeError: If the run does not have an associated build.

    Returns:
        The image name or digest.
    """
    from zenml.client import Client

    run = Client().get_pipeline_run(self.run_id)
    if not run.build:
        raise RuntimeError(
            f"Missing build for run {run.id}. This is probably because "
            "the build was manually deleted."
        )

    return run.build.get_image(
        component_key=key, step=self.pipeline_step_name
    )

store_config

Functionality to support ZenML store configurations.

Classes

StoreConfiguration

Bases: BaseModel

Generic store configuration.

The store configurations of concrete store implementations must inherit from this class and validate any extra attributes that are configured in addition to those defined in this class.

Attributes:

Name	Type	Description
type	StoreType	The type of store backend.
url	str	The URL of the store backend.
secrets_store	Optional[SerializeAsAny[SecretsStoreConfiguration]]	The configuration of the secrets store to use to store secrets. If not set, secrets management is disabled.
backup_secrets_store	Optional[SerializeAsAny[SecretsStoreConfiguration]]	The configuration of the secrets store to use to store backups of secrets. If not set, backup and restore of secrets are disabled.

Functions

supports_url_scheme(url: str) -> bool classmethod

Check if a URL scheme is supported by this store.

Concrete store configuration classes should override this method to check if a URL scheme is supported by the store.

Parameters:

Name	Type	Description	Default
url	str	The URL to check.	required

Returns:

Type	Description
bool	True if the URL scheme is supported, False otherwise.

Source code in src/zenml/config/store_config.py

@classmethod
def supports_url_scheme(cls, url: str) -> bool:
    """Check if a URL scheme is supported by this store.

    Concrete store configuration classes should override this method to
    check if a URL scheme is supported by the store.

    Args:
        url: The URL to check.

    Returns:
        True if the URL scheme is supported, False otherwise.
    """
    return True

validate_store_config(data: Dict[str, Any]) -> Dict[str, Any] classmethod

Validate the secrets store configuration.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	The values of the store configuration.	required

Returns:

Type	Description
Dict[str, Any]	The values of the store configuration.

Source code in src/zenml/config/store_config.py

@model_validator(mode="before")
@classmethod
@before_validator_handler
def validate_store_config(cls, data: Dict[str, Any]) -> Dict[str, Any]:
    """Validate the secrets store configuration.

    Args:
        data: The values of the store configuration.

    Returns:
        The values of the store configuration.
    """
    if data.get("secrets_store") is None:
        return data

    # Remove the legacy REST secrets store configuration since it is no
    # longer supported/needed
    secrets_store = data["secrets_store"]
    if isinstance(secrets_store, dict):
        secrets_store_type = secrets_store.get("type")
        if secrets_store_type == "rest":
            del data["secrets_store"]

    return data

Functions

strict_base_model

Strict immutable pydantic model.

Classes

StrictBaseModel

Bases: BaseModel

Immutable pydantic model which prevents extra attributes.