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

Config

zenml.config special

base_config

BaseConfiguration (ABC)

active_profile: Optional[ProfileConfiguration] property readonly

active_profile_name: Optional[str] property readonly

active_stack_name: Optional[str] property readonly

activate_profile(self, profile_name)

activate_stack(self, stack_name)

config_keys

ConfigKeys

get_keys() classmethod

key_check(config) classmethod

PipelineConfigurationKeys (ConfigKeys)

StepConfigurationKeys (ConfigKeys)

global_config

GlobalConfigMetaClass (ModelMetaclass)

__call__(cls, *args, **kwargs) special

__init__(cls, *args, **kwargs) special

GlobalConfiguration (BaseModel, BaseConfiguration) pydantic-model

active_profile: Optional[zenml.config.profile_config.ProfileConfiguration] property readonly

active_profile_name: Optional[str] property readonly

active_stack_name: Optional[str] property readonly

config_directory: str property readonly

Config

__getattribute__(self, key) special

__init__(self, config_path=None) special

__setattr__(self, key, value) special

activate_profile(self, profile_name)

activate_stack(self, stack_name)

add_or_update_profile(self, profile)

copy_active_configuration(self, config_path, load_config_path=None)

default_config_directory() staticmethod

delete_profile(self, profile_name)

get_instance() classmethod

get_profile(self, profile_name)

has_profile(self, profile_name)

profile_config

ProfileConfiguration (BaseModel) pydantic-model

config_directory: str property readonly

global_config: GlobalConfiguration property readonly

Config

__init__(self, config=None, **kwargs) special

activate_stack(self, stack_name)

activate_user(self, user_name)

cleanup(self)

initialize(self)

get_default_store_type()

`zenml.config` `special`

`base_config`

`BaseConfiguration (ABC)`

`active_profile: Optional[ProfileConfiguration]` `property` `readonly`

`active_profile_name: Optional[str]` `property` `readonly`

`active_stack_name: Optional[str]` `property` `readonly`

`activate_profile(self, profile_name)`

`activate_stack(self, stack_name)`

`config_keys`

`ConfigKeys`

`get_keys()` `classmethod`

`key_check(config)` `classmethod`

`PipelineConfigurationKeys (ConfigKeys)`

`StepConfigurationKeys (ConfigKeys)`

`global_config`

`GlobalConfigMetaClass (ModelMetaclass)`

`call(cls, *args, **kwargs)` `special`

`init(cls, *args, **kwargs)` `special`

`GlobalConfiguration (BaseModel, BaseConfiguration)` `pydantic-model`

`active_profile: Optional[zenml.config.profile_config.ProfileConfiguration]` `property` `readonly`

`active_profile_name: Optional[str]` `property` `readonly`

`active_stack_name: Optional[str]` `property` `readonly`

`config_directory: str` `property` `readonly`

`Config`

`getattribute(self, key)` `special`

`init(self, config_path=None)` `special`

`setattr(self, key, value)` `special`

`activate_profile(self, profile_name)`

`activate_stack(self, stack_name)`

`add_or_update_profile(self, profile)`

`copy_active_configuration(self, config_path, load_config_path=None)`

`default_config_directory()` `staticmethod`

`delete_profile(self, profile_name)`

`get_instance()` `classmethod`

`get_profile(self, profile_name)`

`has_profile(self, profile_name)`

`profile_config`

`ProfileConfiguration (BaseModel)` `pydantic-model`

`config_directory: str` `property` `readonly`

`global_config: GlobalConfiguration` `property` `readonly`

`Config`

`init(self, config=None, **kwargs)` `special`

`activate_stack(self, stack_name)`

`activate_user(self, user_name)`

`cleanup(self)`

`initialize(self)`

`get_default_store_type()`