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
Base class for global configuration management.
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
Validates global configuration values.
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"
SourceConfigurationKeys (ConfigKeys)
Keys for a step configuration source dict.
Source code in zenml/config/config_keys.py
class SourceConfigurationKeys(ConfigKeys):
"""Keys for a step configuration source dict."""
FILE_ = "file"
NAME_ = "name"
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"
docker_configuration
Docker configuration.
DockerConfiguration (BaseModel)
pydantic-model
Configuration for building Docker images to run ZenML pipelines.
Build process:
- No
dockerfilespecified: If any of the configuration options regarding requirements, environment variables or copying files require us to build an image, ZenML will build this image. Otherwise theparent_imagewill be used to run the pipeline. dockerfilespecified: ZenML will first build an image based on the specified Dockerfile. If any of the configuration options regarding requirements, environment variables or copying files require an additional image built on top of that, ZenML will build a second image. If not, the image build from the specified Dockerfile will be used to run the pipeline.
Requirements installation order:
Depending on the configuration of this object, requirements will be
installed in the following order (each step optional):
- The packages installed in your local python environment
- The packages specified via the requirements attribute
- The packages specified via the required_integrations and potentially
stack requirements
Attributes:
| Name | Type | Description |
|---|---|---|
parent_image |
Optional[str] |
Full name of the Docker image that should be used as the parent for the image that will be built. Defaults to a ZenML image built for the active Python and ZenML version. Additional notes:
* If you specify a custom image here, you need to make sure it has
ZenML installed.
* If this is a non-local image, the environment which is running
the pipeline and building the Docker image needs to be able to pull
this image.
* If a custom |
dockerfile |
Optional[str] |
Path to a custom Dockerfile that should be built. Depending on the other values you specify in this configuration, the resulting image will be used directly to run your pipeline or ZenML will use it as a parent image to build on top of. See the general docstring of this class for more information. Additional notes:
* If you specify this, the |
build_context_root |
Optional[str] |
Build context root for the Docker build, only used
when the |
build_options |
Dict[str, Any] |
Additional options that will be passed unmodified to the
Docker build call when building an image using the specified
|
docker_target_repository |
Name of the Docker repository to which the image should be pushed. This repository will be appended to the registry URI of the container registry of your stack and should therefore not include any registry. |
|
replicate_local_python_environment |
Optional[zenml.config.docker_configuration.PythonEnvironmentExportMethod] |
If not |
requirements |
Union[NoneType, str, List[str]] |
Path to a requirements file or a list of required pip
packages. During the image build, these requirements will be
installed using pip. If you need to use a different tool to
resolve and/or install your packages, please use a custom parent
image or specify a custom |
required_integrations |
List[str] |
List of ZenML integrations that should be installed. All requirements for the specified integrations will be installed inside the Docker image. |
install_stack_requirements |
bool |
If |
environment |
Dict[str, Any] |
Dictionary of environment variables to set inside the Docker image. |
dockerignore |
Optional[str] |
Path to a dockerignore file to use when building the Docker image. |
copy_files |
bool |
If |
copy_profile |
bool |
If |
Source code in zenml/config/docker_configuration.py
class DockerConfiguration(BaseModel):
"""Configuration for building Docker images to run ZenML pipelines.
Build process:
--------------
* No `dockerfile` specified: If any of the configuration options regarding
requirements, environment variables or copying files require us to build an
image, ZenML will build this image. Otherwise the `parent_image` will be
used to run the pipeline.
* `dockerfile` specified: ZenML will first build an image based on the
specified Dockerfile. If any of the configuration options regarding
requirements, environment variables or copying files require an additional
image built on top of that, ZenML will build a second image. If not, the
image build from the specified Dockerfile will be used to run the pipeline.
Requirements installation order:
--------------------------------
Depending on the configuration of this object, requirements will be
installed in the following order (each step optional):
- The packages installed in your local python environment
- The packages specified via the `requirements` attribute
- The packages specified via the `required_integrations` and potentially
stack requirements
Attributes:
parent_image: Full name of the Docker image that should be
used as the parent for the image that will be built. Defaults to
a ZenML image built for the active Python and ZenML version.
Additional notes:
* If you specify a custom image here, you need to make sure it has
ZenML installed.
* If this is a non-local image, the environment which is running
the pipeline and building the Docker image needs to be able to pull
this image.
* If a custom `dockerfile` is specified for this configuration
object, this parent image will be ignored.
dockerfile: Path to a custom Dockerfile that should be built. Depending
on the other values you specify in this configuration, the resulting
image will be used directly to run your pipeline or ZenML will use
it as a parent image to build on top of. See the general docstring
of this class for more information.
Additional notes:
* If you specify this, the `parent_image` attribute will be ignored.
* If you specify this, the image built from this Dockerfile needs
to have ZenML installed.
build_context_root: Build context root for the Docker build, only used
when the `dockerfile` attribute is set. If this is left empty, the
build context will only contain the Dockerfile.
build_options: Additional options that will be passed unmodified to the
Docker build call when building an image using the specified
`dockerfile`. You can use this to for example specify build
args or a target stage. See
https://docker-py.readthedocs.io/en/stable/images.html#docker.models.images.ImageCollection.build
for a full list of available options.
docker_target_repository: Name of the Docker repository to which the
image should be pushed. This repository will be appended to the
registry URI of the container registry of your stack and should
therefore **not** include any registry.
replicate_local_python_environment: If not `None`, ZenML will use the
specified method to generate a requirements file that replicates
the packages installed in the currently running python environment.
This requirements file will then be installed in the Docker image.
requirements: Path to a requirements file or a list of required pip
packages. During the image build, these requirements will be
installed using pip. If you need to use a different tool to
resolve and/or install your packages, please use a custom parent
image or specify a custom `dockerfile`.
required_integrations: List of ZenML integrations that should be
installed. All requirements for the specified integrations will
be installed inside the Docker image.
install_stack_requirements: If `True`, ZenML will automatically detect
if components of your active stack are part of a ZenML integration
and install the corresponding requirements. If you set this to
`False` or use custom components in your stack, you need to make
sure these get installed by specifying them in the `requirements`
attribute.
environment: Dictionary of environment variables to set inside the
Docker image.
dockerignore: Path to a dockerignore file to use when building the
Docker image.
copy_files: If `True`, user files will be copied into the Docker image.
If this is set to `False`, ZenML will not copy any of your files
into the Docker image and you're responsible that all the files
to run your pipeline exist in the right place.
copy_profile: If `True`, the configuration of the active profile and
stack will be copied into the Docker image. If this is set to
`False`, ZenML will not copy this configuration and you're
responsible for making sure the right stack is active inside the
Docker image.
"""
parent_image: Optional[str] = None
dockerfile: Optional[str] = None
build_context_root: Optional[str] = None
build_options: Dict[str, Any] = {}
target_repository: str = "zenml"
replicate_local_python_environment: Optional[
PythonEnvironmentExportMethod
] = None
requirements: Union[None, str, List[str]] = None
required_integrations: List[str] = []
install_stack_requirements: bool = True
environment: Dict[str, Any] = {}
dockerignore: Optional[str] = None
copy_files: bool = True
copy_profile: bool = True
class Config:
"""Pydantic configuration class."""
# public attributes are immutable
allow_mutation = False
# prevent extra attributes during model initialization
extra = Extra.forbid
Config
Pydantic configuration class.
Source code in zenml/config/docker_configuration.py
class Config:
"""Pydantic configuration class."""
# public attributes are immutable
allow_mutation = False
# prevent extra attributes during model initialization
extra = Extra.forbid
PythonEnvironmentExportMethod (Enum)
Different methods to export the local Python environment.
Source code in zenml/config/docker_configuration.py
class PythonEnvironmentExportMethod(Enum):
"""Different methods to export the local Python environment."""
PIP_FREEZE = "pip_freeze"
POETRY_EXPORT = "poetry_export"
@property
def command(self) -> str:
"""Shell command that outputs local python packages.
The output string must be something that can be interpreted as a
requirements file for pip once it's written to a file.
Returns:
Shell command.
"""
return {
PythonEnvironmentExportMethod.PIP_FREEZE: "pip freeze",
PythonEnvironmentExportMethod.POETRY_EXPORT: "poetry export --format=requirements.txt",
}[self]
global_config
Functionality to support ZenML GlobalConfiguration.
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.
Args:
*args: positional arguments
**kwargs: keyword arguments
"""
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.
Args:
*args: positional arguments
**kwargs: keyword arguments
Returns:
The global GlobalConfiguration instance.
"""
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
*args |
Any |
positional arguments |
() |
**kwargs |
Any |
keyword arguments |
{} |
Returns:
| Type | Description |
|---|---|
GlobalConfiguration |
The global GlobalConfiguration instance. |
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.
Args:
*args: positional arguments
**kwargs: keyword arguments
Returns:
The global GlobalConfiguration instance.
"""
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
*args |
Any |
positional arguments |
() |
**kwargs |
Any |
keyword arguments |
{} |
Source code in zenml/config/global_config.py
def __init__(cls, *args: Any, **kwargs: Any) -> None:
"""Initialize a singleton class.
Args:
*args: positional arguments
**kwargs: keyword arguments
"""
super().__init__(*args, **kwargs)
cls._global_config: Optional["GlobalConfiguration"] = None
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)
user_metadata: Optional[Dict[str, str]]
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, **kwargs: Any
) -> 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.
**kwargs: keyword arguments
"""
self._config_path = config_path or self.default_config_directory()
config_values = self._read_config()
config_values.update(**kwargs)
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.
Args:
v: The version attribute value.
Returns:
The version attribute value.
Raises:
RuntimeError: If the version parsing fails.
"""
if v is None:
return v
if not isinstance(version.parse(v), version.Version):
# If the version parsing fails, it returns a `LegacyVersion` instead.
# Check to make sure it's an actual `Version` object which represents
# a valid version.
raise RuntimeError(f"Invalid version in global configuration: {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.
Args:
key: The attribute name.
value: The attribute value.
"""
super().__setattr__(key, value)
if key.startswith("_"):
return
self._write_config()
def __custom_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.
Args:
key: The attribute name.
Returns:
The attribute value.
"""
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 Pydantic's 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
if not TYPE_CHECKING:
# When defining __getattribute__, mypy allows accessing non-existent
# attributes without failing
# (see https://github.com/python/mypy/issues/13319).
__getattribute__ = __custom_getattribute__
def _migrate_config(self) -> None:
"""Migrates the global config to the latest version."""
curr_version = version.parse(__version__)
if self.version is None:
logger.info(
"Initializing the ZenML global configuration version to %s",
curr_version,
)
else:
config_version = version.parse(self.version)
if config_version > curr_version:
logger.error(
"The ZenML global configuration version (%s) is higher "
"than the version of ZenML currently being used (%s). "
"This may happen if you recently downgraded ZenML to an "
"earlier version, or if you have already used a more "
"recent ZenML version on the same machine. "
"It is highly recommended that you update ZenML to at "
"least match the global configuration version, otherwise "
"you may run into unexpected issues such as model schema "
"validation failures or even loss of information.",
config_version,
curr_version,
)
# TODO [ENG-899]: Give more detailed instruction on how to resolve
# version mismatch.
return
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.
Returns:
A dictionary containing the configuration options.
"""
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):
io_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.
Returns:
The default global configuration directory.
"""
return io_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.
Returns:
The path to the global configuration file.
"""
return os.path.join(config_path or self._config_path, "config.yaml")
def copy_active_configuration(
self,
config_path: str,
load_config_path: Optional[PurePath] = 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 from `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.
Returns:
A new global configuration object with the active configuration
copied to the specified path.
"""
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
# We don't need to track analytics here
store = Repository.create_store(
profile,
skip_default_registrations=True,
track_analytics=False,
skip_migration=True,
)
# transfer the active stack and all necessary flavors to the new store.
# we disable logs for this call so there is no confusion about newly registered
# stacks/components/flavors
with disable_logging(logging.INFO):
active_stack = repo.zen_store.get_stack(repo.active_stack_name)
store.register_stack(active_stack)
for component in active_stack.components:
try:
flavor = repo.zen_store.get_flavor_by_name_and_type(
flavor_name=component.flavor,
component_type=component.type,
)
store.create_flavor(
source=flavor.source,
name=flavor.name,
stack_component_type=flavor.type,
)
except KeyError:
# not a custom flavor, no need to register anything
pass
# 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:
store_path = store.get_path_from_url(store.url)
if store_path:
relative_store_path = PurePath(store_path).relative_to(
config_copy.config_directory
)
new_store_path = str(load_config_path / relative_store_path)
profile.store_url = store.get_local_url(new_store_path)
else:
# the store url is not a local URL. This only happens
# in case of a REST ZenStore, in which case we don't
# want to modify the URL
profile.store_url = store.url
config_copy._write_config()
return config_copy
@property
def config_directory(self) -> str:
"""Directory where the global configuration file is located.
Returns:
The 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.get_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.
Returns:
| Type | Description |
|---|---|
str |
The 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
__custom_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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key |
str |
The attribute name. |
required |
Returns:
| Type | Description |
|---|---|
Any |
The attribute value. |
Source code in zenml/config/global_config.py
def __custom_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.
Args:
key: The attribute name.
Returns:
The attribute value.
"""
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 Pydantic's 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
__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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key |
str |
The attribute name. |
required |
Returns:
| Type | Description |
|---|---|
Any |
The attribute value. |
Source code in zenml/config/global_config.py
def __custom_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.
Args:
key: The attribute name.
Returns:
The attribute value.
"""
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 Pydantic's 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, **kwargs)
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 |
**kwargs |
Any |
keyword arguments |
{} |
Source code in zenml/config/global_config.py
def __init__(
self, config_path: Optional[str] = None, **kwargs: Any
) -> 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.
**kwargs: keyword arguments
"""
self._config_path = config_path or self.default_config_directory()
config_values = self._read_config()
config_values.update(**kwargs)
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key |
str |
The attribute name. |
required |
value |
Any |
The attribute value. |
required |
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.
Args:
key: The attribute name.
value: The attribute value.
"""
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[pathlib.PurePath] |
path that will be used to load the configuration
copy. This can be set to a value different from |
None |
Returns:
| Type | Description |
|---|---|
GlobalConfiguration |
A new global configuration object with the active configuration copied to the specified path. |
Source code in zenml/config/global_config.py
def copy_active_configuration(
self,
config_path: str,
load_config_path: Optional[PurePath] = 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 from `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.
Returns:
A new global configuration object with the active configuration
copied to the specified path.
"""
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
# We don't need to track analytics here
store = Repository.create_store(
profile,
skip_default_registrations=True,
track_analytics=False,
skip_migration=True,
)
# transfer the active stack and all necessary flavors to the new store.
# we disable logs for this call so there is no confusion about newly registered
# stacks/components/flavors
with disable_logging(logging.INFO):
active_stack = repo.zen_store.get_stack(repo.active_stack_name)
store.register_stack(active_stack)
for component in active_stack.components:
try:
flavor = repo.zen_store.get_flavor_by_name_and_type(
flavor_name=component.flavor,
component_type=component.type,
)
store.create_flavor(
source=flavor.source,
name=flavor.name,
stack_component_type=flavor.type,
)
except KeyError:
# not a custom flavor, no need to register anything
pass
# 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:
store_path = store.get_path_from_url(store.url)
if store_path:
relative_store_path = PurePath(store_path).relative_to(
config_copy.config_directory
)
new_store_path = str(load_config_path / relative_store_path)
profile.store_url = store.get_local_url(new_store_path)
else:
# the store url is not a local URL. This only happens
# in case of a REST ZenStore, in which case we don't
# want to modify the URL
profile.store_url = store.url
config_copy._write_config()
return config_copy
default_config_directory()
staticmethod
Path to the default global configuration directory.
Returns:
| Type | Description |
|---|---|
str |
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.
Returns:
The default global configuration directory.
"""
return io_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
Functionality to support ZenML Profile configuration.
ProfileConfiguration (BaseModel)
pydantic-model
Stores configuration profile options.
Attributes:
| Name | Type | Description |
|---|---|---|
name |
Name of the profile. |
|
active_user |
Name of the active user. |
|
store_url |
URL pointing to the ZenML store backend. |
|
store_type |
Type of the store backend. |
|
active_stack |
Optional name of the active stack. |
|
_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.
active_user: Name of the active user.
store_url: URL pointing to the ZenML store backend.
store_type: Type of the store backend.
active_stack: Optional name of the active stack.
_config: global configuration to which this profile belongs.
"""
name: str
active_user: str
store_url: Optional[str]
store_type: StoreType = Field(default_factory=get_default_store_type)
active_stack: Optional[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.
Returns:
The 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.stack_configurations
except requests.exceptions.ConnectionError:
stacks = None
if stacks:
self.active_stack = list(stacks.keys())[0]
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.
Returns:
The global configuration to which this profile belongs.
"""
from zenml.config.global_config import GlobalConfiguration
return self._config or GlobalConfiguration()
def get_active_stack(self) -> Optional[str]:
"""Get the active stack for the profile.
Returns:
The name of the active stack or None if no stack is active.
"""
return os.getenv(ENV_ZENML_ACTIVATED_STACK, self.active_stack)
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.
Args:
attributes: attributes of the profile configuration
Returns:
attributes of the profile configuration
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']}'."
)
if store_type == StoreType.REST and attributes.get("store_url") is None:
raise RuntimeError(
f"Store URL 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.
Returns:
| Type | Description |
|---|---|
str |
The directory where the profile configuration is stored. |
global_config: GlobalConfiguration
property
readonly
Return the global configuration to which this profile belongs.
Returns:
| Type | Description |
|---|---|
GlobalConfiguration |
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)
get_active_stack(self)
Get the active stack for the profile.
Returns:
| Type | Description |
|---|---|
Optional[str] |
The name of the active stack or None if no stack is active. |
Source code in zenml/config/profile_config.py
def get_active_stack(self) -> Optional[str]:
"""Get the active stack for the profile.
Returns:
The name of the active stack or None if no stack is active.
"""
return os.getenv(ENV_ZENML_ACTIVATED_STACK, self.active_stack)
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.stack_configurations
except requests.exceptions.ConnectionError:
stacks = None
if stacks:
self.active_stack = list(stacks.keys())[0]
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
resource_configuration
Resource Configuration class used to specify resources for a step.
ByteUnit (Enum)
Enum for byte units.
Source code in zenml/config/resource_configuration.py
class ByteUnit(Enum):
"""Enum for byte units."""
KB = "KB"
KIB = "KiB"
MB = "MB"
MIB = "MiB"
GB = "GB"
GIB = "GiB"
TB = "TB"
TIB = "TiB"
PB = "PB"
PIB = "PiB"
@property
def byte_value(self) -> int:
"""Returns the amount of bytes that this unit represents.
Returns:
The byte value of this unit.
"""
return {
ByteUnit.KB: 10**3,
ByteUnit.KIB: 1 << 10,
ByteUnit.MB: 10**6,
ByteUnit.MIB: 1 << 20,
ByteUnit.GB: 10**9,
ByteUnit.GIB: 1 << 30,
ByteUnit.TB: 10**12,
ByteUnit.TIB: 1 << 40,
ByteUnit.PB: 10**15,
ByteUnit.PIB: 1 << 50,
}[self]
ResourceConfiguration (BaseModel)
pydantic-model
Class for configuring hardware resources.
Attributes:
| Name | Type | Description |
|---|---|---|
cpu_count |
Optional[pydantic.types.PositiveFloat] |
The amount of CPU cores that should be configured. |
gpu_count |
Optional[pydantic.types.PositiveInt] |
The amount of GPUs that should be configured. |
memory |
Optional[str] |
The amount of memory that should be configured. |
Source code in zenml/config/resource_configuration.py
class ResourceConfiguration(BaseModel):
"""Class for configuring hardware resources.
Attributes:
cpu_count: The amount of CPU cores that should be configured.
gpu_count: The amount of GPUs that should be configured.
memory: The amount of memory that should be configured.
"""
cpu_count: Optional[PositiveFloat] = None
gpu_count: Optional[PositiveInt] = None
memory: Optional[str] = Field(regex=MEMORY_REGEX)
@property
def empty(self) -> bool:
"""Returns if this object is "empty" (=no values configured) or not.
Returns:
`True` if no values were configured, `False` otherwise.
"""
# To detect whether this config is empty (= no values specified), we
# check if there are any attributes which are explicitly set to any
# value other than `None`.
return len(self.dict(exclude_unset=True, exclude_none=True)) == 0
def get_memory(
self, unit: Union[str, ByteUnit] = ByteUnit.GB
) -> Optional[float]:
"""Gets the memory configuration in a specific unit.
Args:
unit: The unit to which the memory should be converted.
Raises:
ValueError: If the memory string is invalid.
Returns:
The memory configuration converted to the requested unit, or None
if no memory was configured.
"""
if not self.memory:
return None
if isinstance(unit, str):
unit = ByteUnit(unit)
memory = self.memory
for memory_unit in ByteUnit:
if memory.endswith(memory_unit.value):
memory_value = int(memory[: -len(memory_unit.value)])
return memory_value * memory_unit.byte_value / unit.byte_value
else:
# Should never happen due to the regex validation
raise ValueError(f"Unable to parse memory unit from '{memory}'.")
class Config:
"""Pydantic configuration class."""
# public attributes are immutable
allow_mutation = False
# prevent extra attributes during model initialization
extra = Extra.forbid
empty: bool
property
readonly
Returns if this object is "empty" (=no values configured) or not.
Returns:
| Type | Description |
|---|---|
bool |
|
Config
Pydantic configuration class.
Source code in zenml/config/resource_configuration.py
class Config:
"""Pydantic configuration class."""
# public attributes are immutable
allow_mutation = False
# prevent extra attributes during model initialization
extra = Extra.forbid
get_memory(self, unit=<ByteUnit.GB: 'GB'>)
Gets the memory configuration in a specific unit.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
unit |
Union[str, zenml.config.resource_configuration.ByteUnit] |
The unit to which the memory should be converted. |
<ByteUnit.GB: 'GB'> |
Exceptions:
| Type | Description |
|---|---|
ValueError |
If the memory string is invalid. |
Returns:
| Type | Description |
|---|---|
Optional[float] |
The memory configuration converted to the requested unit, or None if no memory was configured. |
Source code in zenml/config/resource_configuration.py
def get_memory(
self, unit: Union[str, ByteUnit] = ByteUnit.GB
) -> Optional[float]:
"""Gets the memory configuration in a specific unit.
Args:
unit: The unit to which the memory should be converted.
Raises:
ValueError: If the memory string is invalid.
Returns:
The memory configuration converted to the requested unit, or None
if no memory was configured.
"""
if not self.memory:
return None
if isinstance(unit, str):
unit = ByteUnit(unit)
memory = self.memory
for memory_unit in ByteUnit:
if memory.endswith(memory_unit.value):
memory_value = int(memory[: -len(memory_unit.value)])
return memory_value * memory_unit.byte_value / unit.byte_value
else:
# Should never happen due to the regex validation
raise ValueError(f"Unable to parse memory unit from '{memory}'.")