Skip to content

Config

zenml.config special

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.

The ProfileConfiguration class is used to model the configuration of a Profile. A GlobalConfiguration object can contain multiple ProfileConfiguration instances.

base_config

BaseConfiguration (ABC)

Base class for global configuration management.

This class defines the common interface related to profile and stack management that all global configuration classes must implement. Both the GlobalConfiguration and Repository classes implement this class, since they share similarities concerning the management of active profiles and stacks.

Source code in zenml/config/base_config.py
class BaseConfiguration(ABC):
    """Base class for global configuration management.

    This class defines the common interface related to profile and stack
    management that all global configuration classes must implement.
    Both the GlobalConfiguration and Repository classes implement this class,
    since they share similarities concerning the management of active profiles
    and stacks.
    """

    @abstractmethod
    def activate_profile(self, profile_name: str) -> None:
        """Set the active profile

        Args:
            profile_name: The name of the profile to set as active.

        Raises:
            KeyError: If the profile with the given name does not exist.
        """

    @property
    @abstractmethod
    def active_profile(self) -> Optional["ProfileConfiguration"]:
        """Return the profile set as active for the repository.

        Returns:
            The active profile or None, if no active profile is set.
        """

    @property
    @abstractmethod
    def active_profile_name(self) -> Optional[str]:
        """Return the name of the profile set as active.

        Returns:
            The active profile name or None, if no active profile is set.
        """

    @abstractmethod
    def activate_stack(self, stack_name: str) -> None:
        """Set the active stack for the active profile.

        Args:
            stack_name: name of the stack to activate

        Raises:
            KeyError: If the stack with the given name does not exist.
        """

    @property
    @abstractmethod
    def active_stack_name(self) -> Optional[str]:
        """Get the active stack name from the active profile.

        Returns:
            The active stack name or None if no active stack is set or if
            no active profile is set.
        """
active_profile: Optional[ProfileConfiguration] property readonly

Return the profile set as active for the repository.

Returns:

Type Description
Optional[ProfileConfiguration]

The active profile or None, if no active profile is set.

active_profile_name: Optional[str] property readonly

Return the name of the profile set as active.

Returns:

Type Description
Optional[str]

The active profile name or None, if no active profile is set.

active_stack_name: Optional[str] property readonly

Get the active stack name from the active profile.

Returns:

Type Description
Optional[str]

The active stack name or None if no active stack is set or if no active profile is set.

activate_profile(self, profile_name)

Set the active profile

Parameters:

Name Type Description Default
profile_name str

The name of the profile to set as active.

required

Exceptions:

Type Description
KeyError

If the profile with the given name does not exist.

Source code in zenml/config/base_config.py
@abstractmethod
def activate_profile(self, profile_name: str) -> None:
    """Set the active profile

    Args:
        profile_name: The name of the profile to set as active.

    Raises:
        KeyError: If the profile with the given name does not exist.
    """
activate_stack(self, stack_name)

Set the active stack for the active profile.

Parameters:

Name Type Description Default
stack_name str

name of the stack to activate

required

Exceptions:

Type Description
KeyError

If the stack with the given name does not exist.

Source code in zenml/config/base_config.py
@abstractmethod
def activate_stack(self, stack_name: str) -> None:
    """Set the active stack for the active profile.

    Args:
        stack_name: name of the stack to activate

    Raises:
        KeyError: If the stack with the given name does not exist.
    """

config_keys

ConfigKeys

Class to validate dictionary configurations.

Source code in zenml/config/config_keys.py
class ConfigKeys:
    """Class to validate dictionary configurations."""

    @classmethod
    def get_keys(cls) -> Tuple[List[str], List[str]]:
        """Gets all the required and optional config keys for this class.

        Returns:
            A tuple (required, optional) which are lists of the
            required/optional keys for this class.
        """
        keys = {
            key: value
            for key, value in cls.__dict__.items()
            if not isinstance(value, classmethod)
            and not isinstance(value, staticmethod)
            and not callable(value)
            and not key.startswith("__")
        }

        required = [v for k, v in keys.items() if not k.endswith("_")]
        optional = [v for k, v in keys.items() if k.endswith("_")]

        return required, optional

    @classmethod
    def key_check(cls, config: Dict[str, Any]) -> None:
        """Checks whether a configuration dict contains all required keys
        and no unknown keys.

        Args:
            config: The configuration dict to verify.

        Raises:
            TypeError: If no config dictionary is passed.
            ValueError: If required keys are missing or unknown keys are found.
        """
        if not isinstance(config, dict):
            raise TypeError(f"Please specify a dict for {cls.__name__}")

        # Required and optional keys for the config dict
        required, optional = cls.get_keys()

        # Check for missing keys
        missing_keys = [k for k in required if k not in config.keys()]
        if missing_keys:
            raise ValueError(f"Missing key(s) {missing_keys} in {cls.__name__}")

        # Check for unknown keys
        unknown_keys = [
            k for k in config.keys() if k not in required and k not in optional
        ]
        if unknown_keys:
            raise ValueError(
                f"Unknown key(s) {unknown_keys} in {cls.__name__}. "
                f"Required keys: {required}, optional keys: {optional}."
            )
get_keys() classmethod

Gets all the required and optional config keys for this class.

Returns:

Type Description
Tuple[List[str], List[str]]

A tuple (required, optional) which are lists of the required/optional keys for this class.

Source code in zenml/config/config_keys.py
@classmethod
def get_keys(cls) -> Tuple[List[str], List[str]]:
    """Gets all the required and optional config keys for this class.

    Returns:
        A tuple (required, optional) which are lists of the
        required/optional keys for this class.
    """
    keys = {
        key: value
        for key, value in cls.__dict__.items()
        if not isinstance(value, classmethod)
        and not isinstance(value, staticmethod)
        and not callable(value)
        and not key.startswith("__")
    }

    required = [v for k, v in keys.items() if not k.endswith("_")]
    optional = [v for k, v in keys.items() if k.endswith("_")]

    return required, optional
key_check(config) classmethod

Checks whether a configuration dict contains all required keys and no unknown keys.

Parameters:

Name Type Description Default
config Dict[str, Any]

The configuration dict to verify.

required

Exceptions:

Type Description
TypeError

If no config dictionary is passed.

ValueError

If required keys are missing or unknown keys are found.

Source code in zenml/config/config_keys.py
@classmethod
def key_check(cls, config: Dict[str, Any]) -> None:
    """Checks whether a configuration dict contains all required keys
    and no unknown keys.

    Args:
        config: The configuration dict to verify.

    Raises:
        TypeError: If no config dictionary is passed.
        ValueError: If required keys are missing or unknown keys are found.
    """
    if not isinstance(config, dict):
        raise TypeError(f"Please specify a dict for {cls.__name__}")

    # Required and optional keys for the config dict
    required, optional = cls.get_keys()

    # Check for missing keys
    missing_keys = [k for k in required if k not in config.keys()]
    if missing_keys:
        raise ValueError(f"Missing key(s) {missing_keys} in {cls.__name__}")

    # Check for unknown keys
    unknown_keys = [
        k for k in config.keys() if k not in required and k not in optional
    ]
    if unknown_keys:
        raise ValueError(
            f"Unknown key(s) {unknown_keys} in {cls.__name__}. "
            f"Required keys: {required}, optional keys: {optional}."
        )

PipelineConfigurationKeys (ConfigKeys)

Keys for a pipeline configuration dict.

Source code in zenml/config/config_keys.py
class PipelineConfigurationKeys(ConfigKeys):
    """Keys for a pipeline configuration dict."""

    NAME = "name"
    STEPS = "steps"

StepConfigurationKeys (ConfigKeys)

Keys for a step configuration dict.

Source code in zenml/config/config_keys.py
class StepConfigurationKeys(ConfigKeys):
    """Keys for a step configuration dict."""

    SOURCE_ = "source"
    PARAMETERS_ = "parameters"
    MATERIALIZERS_ = "materializers"

global_config

GlobalConfigMetaClass (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.
  • an empty default profile is added to the global config on initialization if no other profiles are configured yet.
  • the GlobalConfiguration undergoes a schema migration if the version of the config file is older than the current version of the ZenML package.
Source code in zenml/config/global_config.py
class GlobalConfigMetaClass(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.
    * an empty default profile is added to the global config on initialization
    if no other profiles are configured yet.
    * the GlobalConfiguration undergoes a schema migration if the version of the
    config file is older than the current version of the ZenML package.
    """

    def __init__(cls, *args: Any, **kwargs: Any) -> None:
        """Initialize a singleton class."""
        super().__init__(*args, **kwargs)
        cls._global_config: Optional["GlobalConfiguration"] = None

    def __call__(cls, *args: Any, **kwargs: Any) -> "GlobalConfiguration":
        """Create or return the default global config instance.

        If the GlobalConfiguration constructor is called with custom arguments,
        the singleton functionality of the metaclass is bypassed: a new
        GlobalConfiguration instance is created and returned immediately and
        without saving it as the global GlobalConfiguration singleton.
        """
        if args or kwargs:
            return cast(
                "GlobalConfiguration", super().__call__(*args, **kwargs)
            )

        if not cls._global_config:
            cls._global_config = cast(
                "GlobalConfiguration", super().__call__(*args, **kwargs)
            )
            cls._global_config._migrate_config()
            cls._global_config._add_and_activate_default_profile()

        return cls._global_config
__call__(cls, *args, **kwargs) special

Create or return the default global config instance.

If the GlobalConfiguration constructor is called with custom arguments, the singleton functionality of the metaclass is bypassed: a new GlobalConfiguration instance is created and returned immediately and without saving it as the global GlobalConfiguration singleton.

Source code in zenml/config/global_config.py
def __call__(cls, *args: Any, **kwargs: Any) -> "GlobalConfiguration":
    """Create or return the default global config instance.

    If the GlobalConfiguration constructor is called with custom arguments,
    the singleton functionality of the metaclass is bypassed: a new
    GlobalConfiguration instance is created and returned immediately and
    without saving it as the global GlobalConfiguration singleton.
    """
    if args or kwargs:
        return cast(
            "GlobalConfiguration", super().__call__(*args, **kwargs)
        )

    if not cls._global_config:
        cls._global_config = cast(
            "GlobalConfiguration", super().__call__(*args, **kwargs)
        )
        cls._global_config._migrate_config()
        cls._global_config._add_and_activate_default_profile()

    return cls._global_config
__init__(cls, *args, **kwargs) special

Initialize a singleton class.

Source code in zenml/config/global_config.py
def __init__(cls, *args: Any, **kwargs: Any) -> None:
    """Initialize a singleton class."""
    super().__init__(*args, **kwargs)
    cls._global_config: Optional["GlobalConfiguration"] = None

GlobalConfiguration (BaseModel, BaseConfiguration) pydantic-model

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.

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.

activated_profile Optional[str]

The name of the active configuration profile.

profiles Dict[str, zenml.config.profile_config.ProfileConfiguration]

Map of configuration profiles, indexed by name.

_config_path str

Directory where the global config file is stored.

Source code in zenml/config/global_config.py
class GlobalConfiguration(
    BaseModel, BaseConfiguration, metaclass=GlobalConfigMetaClass
):
    """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:
        user_id: Unique user id.
        analytics_opt_in: If a user agreed to sending analytics or not.
        version: Version of ZenML that was last used to create or update the
            global config.
        activated_profile: The name of the active configuration profile.
        profiles: Map of configuration profiles, indexed by name.
        _config_path: Directory where the global config file is stored.
    """

    user_id: uuid.UUID = Field(default_factory=uuid.uuid4, allow_mutation=False)
    analytics_opt_in: bool = True
    version: Optional[str]
    activated_profile: Optional[str]
    profiles: Dict[str, ProfileConfiguration] = Field(default_factory=dict)
    _config_path: str

    def __init__(self, config_path: Optional[str] = None) -> None:
        """Initializes a GlobalConfiguration object 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 (see the exception below).

        The `config_path` argument is only meant for internal use and testing
        purposes. User code must never pass it to the constructor. When a custom
        `config_path` value is passed, an anonymous GlobalConfiguration instance
        is created and returned independently of the GlobalConfiguration
        singleton and that will have no effect as far as the rest of the ZenML
        core code is concerned.

        If the config file doesn't exist yet, we try to read values from the
        legacy (ZenML version < 0.6) config file.

        Args:
            config_path: (internal use) custom config file path. When not
                specified, the default global configuration path is used and the
                global configuration singleton instance is returned. Only used
                to create configuration copies for transfer to different
                runtime environments.
        """
        self._config_path = config_path or self.default_config_directory()
        config_values = self._read_config()
        super().__init__(**config_values)

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

    @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

    @classmethod
    def _reset_instance(
        cls, config: Optional["GlobalConfiguration"] = None
    ) -> None:
        """Reset the GlobalConfiguration singleton instance.

        This method is only meant for internal use and testing purposes.

        Args:
            config: The GlobalConfiguration instance to set as the global
                singleton. If None, the global GlobalConfiguration singleton is
                reset to an empty value.
        """
        cls._global_config = config

    @validator("version")
    def _validate_version(cls, v: Optional[str]) -> Optional[str]:
        """Validate the version attribute."""
        if v is None:
            return v

        VersionInfo.parse(v)
        return v

    def __setattr__(self, key: str, value: Any) -> None:
        """Sets an attribute on the config and persists the new value in the
        global configuration."""
        super().__setattr__(key, value)
        if key.startswith("_"):
            return
        self._write_config()

    def __getattribute__(self, key: str) -> Any:
        """Gets an attribute value for a specific key.

        If a value for this attribute was specified using an environment
        variable called `$(CONFIG_ENV_VAR_PREFIX)$(ATTRIBUTE_NAME)` and its
        value can be parsed to the attribute type, the value from this
        environment variable is returned instead.
        """
        value = super().__getattribute__(key)
        if key.startswith("_"):
            return value

        environment_variable_name = f"{CONFIG_ENV_VAR_PREFIX}{key.upper()}"
        try:
            environment_variable_value = os.environ[environment_variable_name]
            # set the environment variable value to leverage pydantics type
            # conversion and validation
            super().__setattr__(key, environment_variable_value)
            return_value = super().__getattribute__(key)
            # set back the old value as we don't want to permanently store
            # the environment variable value here
            super().__setattr__(key, value)
            return return_value
        except (ValidationError, KeyError, TypeError):
            return value

    def _migrate_config(self) -> None:
        """Migrates the global config to the latest version."""

        curr_version = VersionInfo.parse(__version__)
        if self.version is None:
            logger.info(
                "Initializing the ZenML global configuration version to %s",
                curr_version,
            )
        else:
            config_version = VersionInfo.parse(self.version)
            if self.version > curr_version:
                raise RuntimeError(
                    "The ZenML global configuration version (%s) is higher "
                    "than the version of ZenML currently being used (%s). "
                    "Please update ZenML to at least match the global "
                    "configuration version to avoid loss of information.",
                    config_version,
                    curr_version,
                )
            if config_version == curr_version:
                return

            logger.info(
                "Migrating the ZenML global configuration from version %s "
                "to version %s...",
                config_version,
                curr_version,
            )

        # this will also trigger rewriting the config file to disk
        # to ensure the schema migration results are persisted
        self.version = __version__

    def _read_config(self) -> Dict[str, Any]:
        """Reads configuration options from disk.

        If the config file doesn't exist yet, this method falls back to reading
        options from a legacy config file or returns an empty dictionary.
        """
        legacy_config_file = os.path.join(
            self.config_directory, LEGACY_CONFIG_FILE_NAME
        )

        config_values = {}
        if fileio.exists(self._config_file()):
            config_values = cast(
                Dict[str, Any],
                yaml_utils.read_yaml(self._config_file()),
            )
        elif fileio.exists(legacy_config_file):
            config_values = cast(
                Dict[str, Any], yaml_utils.read_json(legacy_config_file)
            )

        return config_values

    def _write_config(self, config_path: Optional[str] = None) -> None:
        """Writes the global configuration options to disk.

        Args:
            config_path: custom config file path. When not specified, the default
                global configuration path is used.
        """
        config_file = self._config_file(config_path)
        yaml_dict = json.loads(self.json())
        logger.debug(f"Writing config to {config_file}")

        if not fileio.exists(config_file):
            utils.create_dir_recursive_if_not_exists(
                config_path or self.config_directory
            )

        yaml_utils.write_yaml(config_file, yaml_dict)

    @staticmethod
    def default_config_directory() -> str:
        """Path to the default global configuration directory."""
        return utils.get_global_config_directory()

    def _config_file(self, config_path: Optional[str] = None) -> str:
        """Path to the file where global configuration options are stored.

        Args:
            config_path: custom config file path. When not specified, the default
                global configuration path is used.
        """
        return os.path.join(config_path or self._config_path, "config.yaml")

    def copy_active_configuration(
        self,
        config_path: str,
        load_config_path: Optional[str] = None,
    ) -> "GlobalConfiguration":
        """Create a copy of the global config, the active repository profile
        and the active stack using a different configuration path.

        This method is used to extract the active slice of the current state
        (consisting only of the global configuration, the active profile and the
        active stack) and store it in a different configuration path, where it
        can be loaded in the context of a new environment, such as a container
        image.

        Args:
            config_path: path where the active configuration copy should be saved
            load_config_path: path that will be used to load the configuration
                copy. This can be set to a value different than `config_path`
                if the configuration copy will be loaded from a different
                path, e.g. when the global config copy is copied to a
                container image. This will be reflected in the paths and URLs
                encoded in the profile copy.
        """
        from zenml.repository import Repository

        self._write_config(config_path)

        config_copy = GlobalConfiguration(config_path=config_path)
        config_copy.profiles = {}

        repo = Repository()
        profile = ProfileConfiguration(
            name=repo.active_profile_name,
            active_stack=repo.active_stack_name,
        )

        profile._config = config_copy
        # circumvent the profile initialization done in the
        # ProfileConfiguration and the Repository classes to avoid triggering
        # the analytics and interact directly with the store creation
        config_copy.profiles[profile.name] = profile
        store = Repository.create_store(
            profile, skip_default_registrations=True
        )
        # transfer the active stack to the new store
        store.register_stack(repo.zen_store.get_stack(repo.active_stack_name))

        # if a custom load config path is specified, use it to replace the
        # current store local path in the profile URL
        if load_config_path:
            profile.store_url = store.url.replace(
                str(config_copy.config_directory), load_config_path
            )

        config_copy._write_config()
        return config_copy

    @property
    def config_directory(self) -> str:
        """Directory where the global configuration file is located."""
        return self._config_path

    def add_or_update_profile(
        self, profile: ProfileConfiguration
    ) -> ProfileConfiguration:
        """Adds or updates a profile in the global configuration.

        Args:
            profile: profile configuration

        Returns:
            the profile configuration added to the global configuration
        """
        profile = profile.copy()
        profile._config = self
        if profile.name not in self.profiles:
            profile.initialize()
            track_event(
                AnalyticsEvent.INITIALIZED_PROFILE,
                {"store_type": profile.store_type.value},
            )
        self.profiles[profile.name] = profile
        self._write_config()
        return profile

    def get_profile(self, profile_name: str) -> Optional[ProfileConfiguration]:
        """Get a global configuration profile.

        Args:
            profile_name: name of the profile to get

        Returns:
            The profile configuration or None if the profile doesn't exist
        """
        return self.profiles.get(profile_name)

    def has_profile(self, profile_name: str) -> bool:
        """Check if a named global configuration profile exists.

        Args:
            profile_name: name of the profile to check

        Returns:
            True if the profile exists, otherwise False
        """
        return profile_name in self.profiles

    def activate_profile(self, profile_name: str) -> None:
        """Set a profile as the active.

        Args:
            profile_name: name of the profile to add

        Raises:
            KeyError: If the profile with the given name does not exist.
        """
        if profile_name not in self.profiles:
            raise KeyError(f"Profile '{profile_name}' not found.")
        self.activated_profile = profile_name
        self._write_config()

    def _add_and_activate_default_profile(
        self,
    ) -> Optional[ProfileConfiguration]:
        """Creates and activates the default configuration profile if no
        profiles are configured.

        Returns:
            The newly created default profile or None if other profiles are
            configured.
        """

        if self.profiles:
            return None
        logger.info("Creating default profile...")
        default_profile = ProfileConfiguration(
            name=DEFAULT_PROFILE_NAME,
        )
        self.add_or_update_profile(default_profile)
        self.activate_profile(DEFAULT_PROFILE_NAME)
        logger.info("Created and activated default profile.")

        return default_profile

    @property
    def active_profile(self) -> Optional[ProfileConfiguration]:
        """Return the active profile.

        Returns:
            The active profile.
        """
        if not self.activated_profile:
            return None
        return self.profiles[self.activated_profile]

    @property
    def active_profile_name(self) -> Optional[str]:
        """Return the name of the active profile.

        Returns:
            The name of the active profile.
        """
        return self.activated_profile

    def delete_profile(self, profile_name: str) -> None:
        """Deletes a profile from the global configuration.

        If the profile is active, it cannot be removed.

        Args:
            profile_name: name of the profile to delete

        Raises:
            KeyError: if the profile does not exist
            ValueError: if the profile is active
        """
        if profile_name not in self.profiles:
            raise KeyError(f"Profile '{profile_name}' not found.")
        if profile_name == self.active_profile:
            raise ValueError(
                f"Unable to delete active profile '{profile_name}'."
            )

        profile = self.profiles[profile_name]
        del self.profiles[profile_name]
        profile.cleanup()

        self._write_config()

    def activate_stack(self, stack_name: str) -> None:
        """Set the active stack for the active profile.

        Args:
            stack_name: name of the stack to activate
        """
        if not self.active_profile:
            return
        self.active_profile.active_stack = stack_name
        self._write_config()

    @property
    def active_stack_name(self) -> Optional[str]:
        """Get the active stack name from the active profile.

        Returns:
            The active stack name or None if no active stack is set or if
            no active profile is set.
        """
        if not self.active_profile:
            return None
        return self.active_profile.active_stack

    class Config:
        """Pydantic configuration class."""

        # Validate attributes when assigning them. We need to set this in order
        # to have a mix of mutable and immutable attributes
        validate_assignment = True
        # Ignore extra attributes from configs of previous ZenML versions
        extra = "ignore"
        # all attributes with leading underscore are private and therefore
        # are mutable and not included in serialization
        underscore_attrs_are_private = True
active_profile: Optional[zenml.config.profile_config.ProfileConfiguration] property readonly

Return the active profile.

Returns:

Type Description
Optional[zenml.config.profile_config.ProfileConfiguration]

The active profile.

active_profile_name: Optional[str] property readonly

Return the name of the active profile.

Returns:

Type Description
Optional[str]

The name of the active profile.

active_stack_name: Optional[str] property readonly

Get the active stack name from the active profile.

Returns:

Type Description
Optional[str]

The active stack name or None if no active stack is set or if no active profile is set.

config_directory: str property readonly

Directory where the global configuration file is located.

Config

Pydantic configuration class.

Source code in zenml/config/global_config.py
class Config:
    """Pydantic configuration class."""

    # Validate attributes when assigning them. We need to set this in order
    # to have a mix of mutable and immutable attributes
    validate_assignment = True
    # Ignore extra attributes from configs of previous ZenML versions
    extra = "ignore"
    # all attributes with leading underscore are private and therefore
    # are mutable and not included in serialization
    underscore_attrs_are_private = True
__getattribute__(self, key) special

Gets an attribute value for a specific key.

If a value for this attribute was specified using an environment variable called $(CONFIG_ENV_VAR_PREFIX)$(ATTRIBUTE_NAME) and its value can be parsed to the attribute type, the value from this environment variable is returned instead.

Source code in zenml/config/global_config.py
def __getattribute__(self, key: str) -> Any:
    """Gets an attribute value for a specific key.

    If a value for this attribute was specified using an environment
    variable called `$(CONFIG_ENV_VAR_PREFIX)$(ATTRIBUTE_NAME)` and its
    value can be parsed to the attribute type, the value from this
    environment variable is returned instead.
    """
    value = super().__getattribute__(key)
    if key.startswith("_"):
        return value

    environment_variable_name = f"{CONFIG_ENV_VAR_PREFIX}{key.upper()}"
    try:
        environment_variable_value = os.environ[environment_variable_name]
        # set the environment variable value to leverage pydantics type
        # conversion and validation
        super().__setattr__(key, environment_variable_value)
        return_value = super().__getattribute__(key)
        # set back the old value as we don't want to permanently store
        # the environment variable value here
        super().__setattr__(key, value)
        return return_value
    except (ValidationError, KeyError, TypeError):
        return value
__init__(self, config_path=None) special

Initializes a GlobalConfiguration object 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 (see the exception below).

The config_path argument is only meant for internal use and testing purposes. User code must never pass it to the constructor. When a custom config_path value is passed, an anonymous GlobalConfiguration instance is created and returned independently of the GlobalConfiguration singleton and that will have no effect as far as the rest of the ZenML core code is concerned.

If the config file doesn't exist yet, we try to read values from the legacy (ZenML version < 0.6) config file.

Parameters:

Name Type Description Default
config_path Optional[str]

(internal use) custom config file path. When not specified, the default global configuration path is used and the global configuration singleton instance is returned. Only used to create configuration copies for transfer to different runtime environments.

None
Source code in zenml/config/global_config.py
def __init__(self, config_path: Optional[str] = None) -> None:
    """Initializes a GlobalConfiguration object 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 (see the exception below).

    The `config_path` argument is only meant for internal use and testing
    purposes. User code must never pass it to the constructor. When a custom
    `config_path` value is passed, an anonymous GlobalConfiguration instance
    is created and returned independently of the GlobalConfiguration
    singleton and that will have no effect as far as the rest of the ZenML
    core code is concerned.

    If the config file doesn't exist yet, we try to read values from the
    legacy (ZenML version < 0.6) config file.

    Args:
        config_path: (internal use) custom config file path. When not
            specified, the default global configuration path is used and the
            global configuration singleton instance is returned. Only used
            to create configuration copies for transfer to different
            runtime environments.
    """
    self._config_path = config_path or self.default_config_directory()
    config_values = self._read_config()
    super().__init__(**config_values)

    if not fileio.exists(self._config_file(config_path)):
        self._write_config()
__setattr__(self, key, value) special

Sets an attribute on the config and persists the new value in the global configuration.

Source code in zenml/config/global_config.py
def __setattr__(self, key: str, value: Any) -> None:
    """Sets an attribute on the config and persists the new value in the
    global configuration."""
    super().__setattr__(key, value)
    if key.startswith("_"):
        return
    self._write_config()
activate_profile(self, profile_name)

Set a profile as the active.

Parameters:

Name Type Description Default
profile_name str

name of the profile to add

required

Exceptions:

Type Description
KeyError

If the profile with the given name does not exist.

Source code in zenml/config/global_config.py
def activate_profile(self, profile_name: str) -> None:
    """Set a profile as the active.

    Args:
        profile_name: name of the profile to add

    Raises:
        KeyError: If the profile with the given name does not exist.
    """
    if profile_name not in self.profiles:
        raise KeyError(f"Profile '{profile_name}' not found.")
    self.activated_profile = profile_name
    self._write_config()
activate_stack(self, stack_name)

Set the active stack for the active profile.

Parameters:

Name Type Description Default
stack_name str

name of the stack to activate

required
Source code in zenml/config/global_config.py
def activate_stack(self, stack_name: str) -> None:
    """Set the active stack for the active profile.

    Args:
        stack_name: name of the stack to activate
    """
    if not self.active_profile:
        return
    self.active_profile.active_stack = stack_name
    self._write_config()
add_or_update_profile(self, profile)

Adds or updates a profile in the global configuration.

Parameters:

Name Type Description Default
profile ProfileConfiguration

profile configuration

required

Returns:

Type Description
ProfileConfiguration

the profile configuration added to the global configuration

Source code in zenml/config/global_config.py
def add_or_update_profile(
    self, profile: ProfileConfiguration
) -> ProfileConfiguration:
    """Adds or updates a profile in the global configuration.

    Args:
        profile: profile configuration

    Returns:
        the profile configuration added to the global configuration
    """
    profile = profile.copy()
    profile._config = self
    if profile.name not in self.profiles:
        profile.initialize()
        track_event(
            AnalyticsEvent.INITIALIZED_PROFILE,
            {"store_type": profile.store_type.value},
        )
    self.profiles[profile.name] = profile
    self._write_config()
    return profile
copy_active_configuration(self, config_path, load_config_path=None)

Create a copy of the global config, the active repository profile and the active stack using a different configuration path.

This method is used to extract the active slice of the current state (consisting only of the global configuration, the active profile and the active stack) and store it in a different configuration path, where it can be loaded in the context of a new environment, such as a container image.

Parameters:

Name Type Description Default
config_path str

path where the active configuration copy should be saved

required
load_config_path Optional[str]

path that will be used to load the configuration copy. This can be set to a value different than config_path if the configuration copy will be loaded from a different path, e.g. when the global config copy is copied to a container image. This will be reflected in the paths and URLs encoded in the profile copy.

None
Source code in zenml/config/global_config.py
def copy_active_configuration(
    self,
    config_path: str,
    load_config_path: Optional[str] = None,
) -> "GlobalConfiguration":
    """Create a copy of the global config, the active repository profile
    and the active stack using a different configuration path.

    This method is used to extract the active slice of the current state
    (consisting only of the global configuration, the active profile and the
    active stack) and store it in a different configuration path, where it
    can be loaded in the context of a new environment, such as a container
    image.

    Args:
        config_path: path where the active configuration copy should be saved
        load_config_path: path that will be used to load the configuration
            copy. This can be set to a value different than `config_path`
            if the configuration copy will be loaded from a different
            path, e.g. when the global config copy is copied to a
            container image. This will be reflected in the paths and URLs
            encoded in the profile copy.
    """
    from zenml.repository import Repository

    self._write_config(config_path)

    config_copy = GlobalConfiguration(config_path=config_path)
    config_copy.profiles = {}

    repo = Repository()
    profile = ProfileConfiguration(
        name=repo.active_profile_name,
        active_stack=repo.active_stack_name,
    )

    profile._config = config_copy
    # circumvent the profile initialization done in the
    # ProfileConfiguration and the Repository classes to avoid triggering
    # the analytics and interact directly with the store creation
    config_copy.profiles[profile.name] = profile
    store = Repository.create_store(
        profile, skip_default_registrations=True
    )
    # transfer the active stack to the new store
    store.register_stack(repo.zen_store.get_stack(repo.active_stack_name))

    # if a custom load config path is specified, use it to replace the
    # current store local path in the profile URL
    if load_config_path:
        profile.store_url = store.url.replace(
            str(config_copy.config_directory), load_config_path
        )

    config_copy._write_config()
    return config_copy
default_config_directory() staticmethod

Path to the default global configuration directory.

Source code in zenml/config/global_config.py
@staticmethod
def default_config_directory() -> str:
    """Path to the default global configuration directory."""
    return utils.get_global_config_directory()
delete_profile(self, profile_name)

Deletes a profile from the global configuration.

If the profile is active, it cannot be removed.

Parameters:

Name Type Description Default
profile_name str

name of the profile to delete

required

Exceptions:

Type Description
KeyError

if the profile does not exist

ValueError

if the profile is active

Source code in zenml/config/global_config.py
def delete_profile(self, profile_name: str) -> None:
    """Deletes a profile from the global configuration.

    If the profile is active, it cannot be removed.

    Args:
        profile_name: name of the profile to delete

    Raises:
        KeyError: if the profile does not exist
        ValueError: if the profile is active
    """
    if profile_name not in self.profiles:
        raise KeyError(f"Profile '{profile_name}' not found.")
    if profile_name == self.active_profile:
        raise ValueError(
            f"Unable to delete active profile '{profile_name}'."
        )

    profile = self.profiles[profile_name]
    del self.profiles[profile_name]
    profile.cleanup()

    self._write_config()
get_instance() classmethod

Return the GlobalConfiguration singleton instance.

Returns:

Type Description
Optional[GlobalConfiguration]

The GlobalConfiguration singleton instance or None, if the GlobalConfiguration hasn't been initialized yet.

Source code in 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
get_profile(self, profile_name)

Get a global configuration profile.

Parameters:

Name Type Description Default
profile_name str

name of the profile to get

required

Returns:

Type Description
Optional[zenml.config.profile_config.ProfileConfiguration]

The profile configuration or None if the profile doesn't exist

Source code in zenml/config/global_config.py
def get_profile(self, profile_name: str) -> Optional[ProfileConfiguration]:
    """Get a global configuration profile.

    Args:
        profile_name: name of the profile to get

    Returns:
        The profile configuration or None if the profile doesn't exist
    """
    return self.profiles.get(profile_name)
has_profile(self, profile_name)

Check if a named global configuration profile exists.

Parameters:

Name Type Description Default
profile_name str

name of the profile to check

required

Returns:

Type Description
bool

True if the profile exists, otherwise False

Source code in zenml/config/global_config.py
def has_profile(self, profile_name: str) -> bool:
    """Check if a named global configuration profile exists.

    Args:
        profile_name: name of the profile to check

    Returns:
        True if the profile exists, otherwise False
    """
    return profile_name in self.profiles

profile_config

ProfileConfiguration (BaseModel) pydantic-model

Stores configuration profile options.

Attributes:

Name Type Description
name

Name of the profile.

store_url

URL pointing to the ZenML store backend.

store_type

Type of the store backend.

active_stack

Optional name of the active stack.

active_user

Name of the active user.

_config

global configuration to which this profile belongs.

Source code in zenml/config/profile_config.py
class ProfileConfiguration(BaseModel):
    """Stores configuration profile options.

    Attributes:
        name: Name of the profile.
        store_url: URL pointing to the ZenML store backend.
        store_type: Type of the store backend.
        active_stack: Optional name of the active stack.
        active_user: Name of the active user.
        _config: global configuration to which this profile belongs.
    """

    name: str
    store_url: Optional[str]
    store_type: StoreType = Field(default_factory=get_default_store_type)
    active_stack: Optional[str]
    active_user: str
    _config: Optional["GlobalConfiguration"]

    def __init__(
        self, config: Optional["GlobalConfiguration"] = None, **kwargs: Any
    ) -> None:
        """Initializes a ProfileConfiguration object.

        Args:
            config: global configuration to which this profile belongs. When not
                specified, the default global configuration path is used.
            **kwargs: additional keyword arguments are passed to the
                BaseModel constructor.
        """
        self._config = config
        super().__init__(**kwargs)

    @property
    def config_directory(self) -> str:
        """Directory where the profile configuration is stored."""
        return os.path.join(
            self.global_config.config_directory, "profiles", self.name
        )

    def initialize(self) -> None:
        """Initialize the profile."""

        # import here to avoid circular dependency
        from zenml.repository import Repository

        logger.info("Initializing profile `%s`...", self.name)

        # Create and initialize the profile using a special repository instance.
        # This also validates and updates the store URL configuration and
        # creates all necessary resources (e.g. paths, initial DB, default
        # stacks).
        repo = Repository(profile=self)

        if not self.active_stack:
            try:
                stacks = repo.stacks
            except requests.exceptions.ConnectionError:
                stacks = None
            if stacks:
                self.active_stack = stacks[0].name

    def cleanup(self) -> None:
        """Cleanup the profile directory."""
        if fileio.isdir(self.config_directory):
            fileio.rmtree(self.config_directory)

    @property
    def global_config(self) -> "GlobalConfiguration":
        """Return the global configuration to which this profile belongs."""
        from zenml.config.global_config import GlobalConfiguration

        return self._config or GlobalConfiguration()

    def activate_stack(self, stack_name: str) -> None:
        """Set the active stack for the profile.

        Args:
            stack_name: name of the stack to activate
        """
        self.active_stack = stack_name
        self.global_config._write_config()

    def activate_user(self, user_name: str) -> None:
        """Set the active user for the profile.

        Args:
            user_name: name of the user to activate
        """
        self.active_user = user_name
        self.global_config._write_config()

    @root_validator(pre=True)
    def _ensure_active_user_is_set(
        cls, attributes: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Ensures that an active user is set for this profile.

        If the active user is missing and the profile specifies a local store,
        a default user is used as fallback.

        Raises:
            RuntimeError: If the active user is missing for a profile with a
                REST ZenStore.
        """
        store_type = attributes.get("store_type") or get_default_store_type()

        if (
            store_type != StoreType.REST
            and attributes.get("active_user") is None
        ):
            # in case of a local store, fallback to the default user that is
            # created when initializing the store
            from zenml.zen_stores.base_zen_store import DEFAULT_USERNAME

            attributes["active_user"] = DEFAULT_USERNAME

        if not attributes.get("active_user"):
            raise RuntimeError(
                f"Active user missing for profile '{attributes['name']}'."
            )

        return attributes

    class Config:
        """Pydantic configuration class."""

        # Validate attributes when assigning them. We need to set this in order
        # to have a mix of mutable and immutable attributes
        validate_assignment = True
        # Ignore extra attributes from configs of previous ZenML versions
        extra = "ignore"
        # all attributes with leading underscore are private and therefore
        # are mutable and not included in serialization
        underscore_attrs_are_private = True
config_directory: str property readonly

Directory where the profile configuration is stored.

global_config: GlobalConfiguration property readonly

Return the global configuration to which this profile belongs.

Config

Pydantic configuration class.

Source code in zenml/config/profile_config.py
class Config:
    """Pydantic configuration class."""

    # Validate attributes when assigning them. We need to set this in order
    # to have a mix of mutable and immutable attributes
    validate_assignment = True
    # Ignore extra attributes from configs of previous ZenML versions
    extra = "ignore"
    # all attributes with leading underscore are private and therefore
    # are mutable and not included in serialization
    underscore_attrs_are_private = True
__init__(self, config=None, **kwargs) special

Initializes a ProfileConfiguration object.

Parameters:

Name Type Description Default
config Optional[GlobalConfiguration]

global configuration to which this profile belongs. When not specified, the default global configuration path is used.

None
**kwargs Any

additional keyword arguments are passed to the BaseModel constructor.

{}
Source code in zenml/config/profile_config.py
def __init__(
    self, config: Optional["GlobalConfiguration"] = None, **kwargs: Any
) -> None:
    """Initializes a ProfileConfiguration object.

    Args:
        config: global configuration to which this profile belongs. When not
            specified, the default global configuration path is used.
        **kwargs: additional keyword arguments are passed to the
            BaseModel constructor.
    """
    self._config = config
    super().__init__(**kwargs)
activate_stack(self, stack_name)

Set the active stack for the profile.

Parameters:

Name Type Description Default
stack_name str

name of the stack to activate

required
Source code in zenml/config/profile_config.py
def activate_stack(self, stack_name: str) -> None:
    """Set the active stack for the profile.

    Args:
        stack_name: name of the stack to activate
    """
    self.active_stack = stack_name
    self.global_config._write_config()
activate_user(self, user_name)

Set the active user for the profile.

Parameters:

Name Type Description Default
user_name str

name of the user to activate

required
Source code in zenml/config/profile_config.py
def activate_user(self, user_name: str) -> None:
    """Set the active user for the profile.

    Args:
        user_name: name of the user to activate
    """
    self.active_user = user_name
    self.global_config._write_config()
cleanup(self)

Cleanup the profile directory.

Source code in zenml/config/profile_config.py
def cleanup(self) -> None:
    """Cleanup the profile directory."""
    if fileio.isdir(self.config_directory):
        fileio.rmtree(self.config_directory)
initialize(self)

Initialize the profile.

Source code in zenml/config/profile_config.py
def initialize(self) -> None:
    """Initialize the profile."""

    # import here to avoid circular dependency
    from zenml.repository import Repository

    logger.info("Initializing profile `%s`...", self.name)

    # Create and initialize the profile using a special repository instance.
    # This also validates and updates the store URL configuration and
    # creates all necessary resources (e.g. paths, initial DB, default
    # stacks).
    repo = Repository(profile=self)

    if not self.active_stack:
        try:
            stacks = repo.stacks
        except requests.exceptions.ConnectionError:
            stacks = None
        if stacks:
            self.active_stack = stacks[0].name

get_default_store_type()

Return the default store type.

The default store type can be set via the environment variable ZENML_DEFAULT_STORE_TYPE. If this variable is not set, the default store type is set to 'LOCAL'.

NOTE: this is a global function instead of a default ProfileConfiguration.store_type value because it makes it easier to mock in the unit tests.

Returns:

Type Description
StoreType

The default store type.

Source code in zenml/config/profile_config.py
def get_default_store_type() -> StoreType:
    """Return the default store type.

    The default store type can be set via the environment variable
    ZENML_DEFAULT_STORE_TYPE. If this variable is not set, the default
    store type is set to 'LOCAL'.

    NOTE: this is a global function instead of a default
    `ProfileConfiguration.store_type` value because it makes it easier to mock
    in the unit tests.

    Returns:
        The default store type.
    """
    store_type = os.getenv(ENV_ZENML_DEFAULT_STORE_TYPE)
    if store_type and store_type in StoreType.values():
        return StoreType(store_type)
    return StoreType.LOCAL