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 .zenglobal.json, located on the user's directory for configuration files. (The exact location differs from operating system to operating system.)

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

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:
            AssertionError: If the dictionary contains unknown keys or
                is missing any required key.
        """
        assert isinstance(config, dict), "Please specify a dict for {}".format(
            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()]
        assert len(missing_keys) == 0, "Missing key(s) {} in {}".format(
            missing_keys, cls.__name__
        )

        # Check for unknown keys
        unknown_keys = [
            k for k in config.keys() if k not in required and k not in optional
        ]
        assert (
            len(unknown_keys) == 0
        ), "Unknown key(s) {} in {}. Required keys : {} " "Optional Keys: {}".format(
            unknown_keys,
            cls.__name__,
            required,
            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
AssertionError	If the dictionary contains unknown keys or is missing any required key.

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:
        AssertionError: If the dictionary contains unknown keys or
            is missing any required key.
    """
    assert isinstance(config, dict), "Please specify a dict for {}".format(
        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()]
    assert len(missing_keys) == 0, "Missing key(s) {} in {}".format(
        missing_keys, cls.__name__
    )

    # Check for unknown keys
    unknown_keys = [
        k for k in config.keys() if k not in required and k not in optional
    ]
    assert (
        len(unknown_keys) == 0
    ), "Unknown key(s) {} in {}. Required keys : {} " "Optional Keys: {}".format(
        unknown_keys,
        cls.__name__,
        required,
        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

GlobalConfig (BaseModel) pydantic-model

Stores global configuration options.

Configuration options are read from a config file, but can be overwritten by environment variables. See GlobalConfig.__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.

Source code in zenml/config/global_config.py

class GlobalConfig(BaseModel):
    """Stores global configuration options.

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

    Attributes:
        user_id: Unique user id.
        analytics_opt_in: If a user agreed to sending analytics or not.
    """

    user_id: uuid.UUID = Field(default_factory=uuid.uuid4, allow_mutation=False)
    analytics_opt_in: bool = True

    def __init__(self) -> None:
        """Initializes a GlobalConfig object using values from the config file.

        If the config file doesn't exist yet, we try to read values from the
        legacy (ZenML version < 0.6) config file.
        """
        config_values = self._read_config()
        super().__init__(**config_values)

        if not fileio.file_exists(self.config_file()):
            # the config file hasn't been written to disk, make sure to persist
            # the unique user id
            fileio.create_dir_recursive_if_not_exists(self.config_directory())
            self._write_config()

    def __setattr__(self, key: str, value: Any) -> None:
        """Sets an attribute on the global config and persists the new value."""
        super().__setattr__(key, value)
        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 `ZENML_$(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)

        environment_variable_name = f"ZENML_{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 _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(
            GlobalConfig.config_directory(), LEGACY_CONFIG_FILE_NAME
        )

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

        return config_values

    def _write_config(self) -> None:
        """Writes the global configuration options to disk."""
        yaml_dict = json.loads(self.json())
        yaml_utils.write_yaml(self.config_file(), yaml_dict)

    @staticmethod
    def config_directory() -> str:
        """Path to the global configuration directory."""
        # TODO [ENG-370]: Remove the util method to get global config directory,
        #  the remaining codebase should use `GlobalConfig.config_directory()`
        #  instead.
        return get_global_config_directory()

    @staticmethod
    def config_file() -> str:
        """Path to the file where global configuration options are stored."""
        return os.path.join(GlobalConfig.config_directory(), "config.yaml")

    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"

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"

__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 ZENML_$(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 `ZENML_$(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)

    environment_variable_name = f"ZENML_{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) special

Initializes a GlobalConfig object using values from the config file.

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

Source code in zenml/config/global_config.py

def __init__(self) -> None:
    """Initializes a GlobalConfig object using values from the config file.

    If the config file doesn't exist yet, we try to read values from the
    legacy (ZenML version < 0.6) config file.
    """
    config_values = self._read_config()
    super().__init__(**config_values)

    if not fileio.file_exists(self.config_file()):
        # the config file hasn't been written to disk, make sure to persist
        # the unique user id
        fileio.create_dir_recursive_if_not_exists(self.config_directory())
        self._write_config()

__setattr__(self, key, value) special

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

Source code in zenml/config/global_config.py

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

config_directory() staticmethod

Path to the global configuration directory.

Source code in zenml/config/global_config.py

@staticmethod
def config_directory() -> str:
    """Path to the global configuration directory."""
    # TODO [ENG-370]: Remove the util method to get global config directory,
    #  the remaining codebase should use `GlobalConfig.config_directory()`
    #  instead.
    return get_global_config_directory()

config_file() staticmethod

Path to the file where global configuration options are stored.

Source code in zenml/config/global_config.py

@staticmethod
def config_file() -> str:
    """Path to the file where global configuration options are stored."""
    return os.path.join(GlobalConfig.config_directory(), "config.yaml")