Container Registries
zenml.container_registries
special
Initialization for ZenML's container registries module.
A container registry is a store for (Docker) containers. A ZenML workflow involving a container registry would automatically containerize your code to be transported across stacks running remotely. As part of the deployment to the cluster, the ZenML base image would be downloaded (from a cloud container registry) and used as the basis for the deployed 'run'.
For instance, when you are running a local container-based stack, you would therefore have a local container registry which stores the container images you create that bundle up your pipeline code. You could also use a remote container registry like the Elastic Container Registry at AWS in a more production setting.
azure_container_registry
Implementation of an Azure Container Registry class.
AzureContainerRegistryFlavor (BaseContainerRegistryFlavor)
Class for Azure Container Registry.
Source code in zenml/container_registries/azure_container_registry.py
class AzureContainerRegistryFlavor(BaseContainerRegistryFlavor):
"""Class for Azure Container Registry."""
@property
def name(self) -> str:
"""Name of the flavor.
Returns:
The name of the flavor.
"""
return ContainerRegistryFlavor.AZURE.value
@property
def service_connector_requirements(
self,
) -> Optional[ServiceConnectorRequirements]:
"""Service connector resource requirements for service connectors.
Specifies resource requirements that are used to filter the available
service connector types that are compatible with this flavor.
Returns:
Requirements for compatible service connectors, if a service
connector is required for this flavor.
"""
return ServiceConnectorRequirements(
connector_type="azure",
resource_type=DOCKER_REGISTRY_RESOURCE_TYPE,
resource_id_attr="uri",
)
@property
def docs_url(self) -> Optional[str]:
"""A url to point at docs explaining this flavor.
Returns:
A flavor docs url.
"""
return self.generate_default_docs_url()
@property
def sdk_docs_url(self) -> Optional[str]:
"""A url to point at docs explaining this flavor.
Returns:
A flavor docs url.
"""
return self.generate_default_sdk_docs_url()
@property
def logo_url(self) -> str:
"""A url to represent the flavor in the dashboard.
Returns:
The flavor logo.
"""
return "https://public-flavor-logos.s3.eu-central-1.amazonaws.com/container_registry/azure.png"
docs_url: Optional[str]
property
readonly
A url to point at docs explaining this flavor.
Returns:
Type | Description |
---|---|
Optional[str] |
A flavor docs url. |
logo_url: str
property
readonly
A url to represent the flavor in the dashboard.
Returns:
Type | Description |
---|---|
str |
The flavor logo. |
name: str
property
readonly
Name of the flavor.
Returns:
Type | Description |
---|---|
str |
The name of the flavor. |
sdk_docs_url: Optional[str]
property
readonly
A url to point at docs explaining this flavor.
Returns:
Type | Description |
---|---|
Optional[str] |
A flavor docs url. |
service_connector_requirements: Optional[zenml.models.v2.misc.service_connector_type.ServiceConnectorRequirements]
property
readonly
Service connector resource requirements for service connectors.
Specifies resource requirements that are used to filter the available service connector types that are compatible with this flavor.
Returns:
Type | Description |
---|---|
Optional[zenml.models.v2.misc.service_connector_type.ServiceConnectorRequirements] |
Requirements for compatible service connectors, if a service connector is required for this flavor. |
base_container_registry
Implementation of a base container registry class.
BaseContainerRegistry (AuthenticationMixin)
Base class for all ZenML container registries.
Source code in zenml/container_registries/base_container_registry.py
class BaseContainerRegistry(AuthenticationMixin):
"""Base class for all ZenML container registries."""
_docker_client: Optional["DockerClient"] = None
@property
def config(self) -> BaseContainerRegistryConfig:
"""Returns the `BaseContainerRegistryConfig` config.
Returns:
The configuration.
"""
return cast(BaseContainerRegistryConfig, self._config)
@property
def requires_authentication(self) -> bool:
"""Returns whether the container registry requires authentication.
Returns:
`True` if the container registry requires authentication,
`False` otherwise.
"""
return bool(self.config.authentication_secret)
@property
def credentials(self) -> Optional[Tuple[str, str]]:
"""Username and password to authenticate with this container registry.
Returns:
Tuple with username and password if this container registry
requires authentication, `None` otherwise.
"""
secret = self.get_typed_authentication_secret(
expected_schema_type=BasicAuthSecretSchema
)
if secret:
return secret.username, secret.password
connector = self.get_connector()
if connector:
from zenml.service_connectors.docker_service_connector import (
DockerServiceConnector,
)
if isinstance(connector, DockerServiceConnector):
return (
connector.config.username.get_secret_value(),
connector.config.password.get_secret_value(),
)
return None
@property
def docker_client(self) -> "DockerClient":
"""Returns a Docker client for this container registry.
Returns:
The Docker client.
Raises:
RuntimeError: If the connector does not return a Docker client.
"""
from docker.client import DockerClient
# Refresh the client also if the connector has expired
if self._docker_client and not self.connector_has_expired():
return self._docker_client
connector = self.get_connector()
if connector:
client = connector.connect()
if not isinstance(client, DockerClient):
raise RuntimeError(
f"Expected a DockerClient while trying to use the "
f"linked connector, but got {type(client)}."
)
self._docker_client = client
else:
self._docker_client = (
docker_utils._try_get_docker_client_from_env()
)
credentials = self.credentials
if credentials:
username, password = credentials
self._docker_client.login(
username=username,
password=password,
registry=self.config.uri,
reauth=True,
)
return self._docker_client
def prepare_image_push(self, image_name: str) -> None:
"""Preparation before an image gets pushed.
Subclasses can overwrite this to do any necessary checks or
preparations before an image gets pushed.
Args:
image_name: Name of the docker image that will be pushed.
"""
def push_image(self, image_name: str) -> str:
"""Pushes a docker image.
Args:
image_name: Name of the docker image that will be pushed.
Returns:
The Docker repository digest of the pushed image.
Raises:
ValueError: If the image name is not associated with this
container registry.
"""
if not image_name.startswith(self.config.uri):
raise ValueError(
f"Docker image `{image_name}` does not belong to container "
f"registry `{self.config.uri}`."
)
self.prepare_image_push(image_name)
return docker_utils.push_image(
image_name, docker_client=self.docker_client
)
config: BaseContainerRegistryConfig
property
readonly
Returns the BaseContainerRegistryConfig
config.
Returns:
Type | Description |
---|---|
BaseContainerRegistryConfig |
The configuration. |
credentials: Optional[Tuple[str, str]]
property
readonly
Username and password to authenticate with this container registry.
Returns:
Type | Description |
---|---|
Optional[Tuple[str, str]] |
Tuple with username and password if this container registry
requires authentication, |
docker_client: DockerClient
property
readonly
Returns a Docker client for this container registry.
Returns:
Type | Description |
---|---|
DockerClient |
The Docker client. |
Exceptions:
Type | Description |
---|---|
RuntimeError |
If the connector does not return a Docker client. |
requires_authentication: bool
property
readonly
Returns whether the container registry requires authentication.
Returns:
Type | Description |
---|---|
bool |
|
prepare_image_push(self, image_name)
Preparation before an image gets pushed.
Subclasses can overwrite this to do any necessary checks or preparations before an image gets pushed.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
image_name |
str |
Name of the docker image that will be pushed. |
required |
Source code in zenml/container_registries/base_container_registry.py
def prepare_image_push(self, image_name: str) -> None:
"""Preparation before an image gets pushed.
Subclasses can overwrite this to do any necessary checks or
preparations before an image gets pushed.
Args:
image_name: Name of the docker image that will be pushed.
"""
push_image(self, image_name)
Pushes a docker image.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
image_name |
str |
Name of the docker image that will be pushed. |
required |
Returns:
Type | Description |
---|---|
str |
The Docker repository digest of the pushed image. |
Exceptions:
Type | Description |
---|---|
ValueError |
If the image name is not associated with this container registry. |
Source code in zenml/container_registries/base_container_registry.py
def push_image(self, image_name: str) -> str:
"""Pushes a docker image.
Args:
image_name: Name of the docker image that will be pushed.
Returns:
The Docker repository digest of the pushed image.
Raises:
ValueError: If the image name is not associated with this
container registry.
"""
if not image_name.startswith(self.config.uri):
raise ValueError(
f"Docker image `{image_name}` does not belong to container "
f"registry `{self.config.uri}`."
)
self.prepare_image_push(image_name)
return docker_utils.push_image(
image_name, docker_client=self.docker_client
)
BaseContainerRegistryConfig (AuthenticationConfigMixin)
Base config for a container registry.
Attributes:
Name | Type | Description |
---|---|---|
uri |
str |
The URI of the container registry. |
Source code in zenml/container_registries/base_container_registry.py
class BaseContainerRegistryConfig(AuthenticationConfigMixin):
"""Base config for a container registry.
Attributes:
uri: The URI of the container registry.
"""
uri: str
@field_validator("uri")
@classmethod
def strip_trailing_slash(cls, uri: str) -> str:
"""Removes trailing slashes from the URI.
Args:
uri: The URI to be stripped.
Returns:
The URI without trailing slashes.
"""
return uri.rstrip("/")
@property
def is_local(self) -> bool:
"""Checks if this stack component is running locally.
Returns:
True if this config is for a local component, False otherwise.
"""
return bool(re.fullmatch(r"localhost:[0-9]{4,5}", self.uri))
is_local: bool
property
readonly
Checks if this stack component is running locally.
Returns:
Type | Description |
---|---|
bool |
True if this config is for a local component, False otherwise. |
strip_trailing_slash(uri)
classmethod
Removes trailing slashes from the URI.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
uri |
str |
The URI to be stripped. |
required |
Returns:
Type | Description |
---|---|
str |
The URI without trailing slashes. |
Source code in zenml/container_registries/base_container_registry.py
@field_validator("uri")
@classmethod
def strip_trailing_slash(cls, uri: str) -> str:
"""Removes trailing slashes from the URI.
Args:
uri: The URI to be stripped.
Returns:
The URI without trailing slashes.
"""
return uri.rstrip("/")
BaseContainerRegistryFlavor (Flavor)
Base flavor for container registries.
Source code in zenml/container_registries/base_container_registry.py
class BaseContainerRegistryFlavor(Flavor):
"""Base flavor for container registries."""
@property
def type(self) -> StackComponentType:
"""Returns the flavor type.
Returns:
The flavor type.
"""
return StackComponentType.CONTAINER_REGISTRY
@property
def service_connector_requirements(
self,
) -> Optional[ServiceConnectorRequirements]:
"""Service connector resource requirements for service connectors.
Specifies resource requirements that are used to filter the available
service connector types that are compatible with this flavor.
Returns:
Requirements for compatible service connectors, if a service
connector is required for this flavor.
"""
return ServiceConnectorRequirements(
resource_type=DOCKER_REGISTRY_RESOURCE_TYPE,
resource_id_attr="uri",
)
@property
def config_class(self) -> Type[BaseContainerRegistryConfig]:
"""Config class for this flavor.
Returns:
The config class.
"""
return BaseContainerRegistryConfig
@property
def implementation_class(self) -> Type[BaseContainerRegistry]:
"""Implementation class.
Returns:
The implementation class.
"""
return BaseContainerRegistry
config_class: Type[zenml.container_registries.base_container_registry.BaseContainerRegistryConfig]
property
readonly
Config class for this flavor.
Returns:
Type | Description |
---|---|
Type[zenml.container_registries.base_container_registry.BaseContainerRegistryConfig] |
The config class. |
implementation_class: Type[zenml.container_registries.base_container_registry.BaseContainerRegistry]
property
readonly
Implementation class.
Returns:
Type | Description |
---|---|
Type[zenml.container_registries.base_container_registry.BaseContainerRegistry] |
The implementation class. |
service_connector_requirements: Optional[zenml.models.v2.misc.service_connector_type.ServiceConnectorRequirements]
property
readonly
Service connector resource requirements for service connectors.
Specifies resource requirements that are used to filter the available service connector types that are compatible with this flavor.
Returns:
Type | Description |
---|---|
Optional[zenml.models.v2.misc.service_connector_type.ServiceConnectorRequirements] |
Requirements for compatible service connectors, if a service connector is required for this flavor. |
type: StackComponentType
property
readonly
Returns the flavor type.
Returns:
Type | Description |
---|---|
StackComponentType |
The flavor type. |
default_container_registry
Implementation of a default container registry class.
DefaultContainerRegistryFlavor (BaseContainerRegistryFlavor)
Class for default ZenML container registries.
Source code in zenml/container_registries/default_container_registry.py
class DefaultContainerRegistryFlavor(BaseContainerRegistryFlavor):
"""Class for default ZenML container registries."""
@property
def name(self) -> str:
"""Name of the flavor.
Returns:
The name of the flavor.
"""
return ContainerRegistryFlavor.DEFAULT.value
@property
def docs_url(self) -> Optional[str]:
"""A URL to point at docs explaining this flavor.
Returns:
A flavor docs url.
"""
return self.generate_default_docs_url()
@property
def sdk_docs_url(self) -> Optional[str]:
"""A URL to point at docs explaining this flavor.
Returns:
A flavor docs url.
"""
return self.generate_default_sdk_docs_url()
@property
def logo_url(self) -> str:
"""A URL to represent the flavor in the dashboard.
Returns:
The flavor logo.
"""
return "https://public-flavor-logos.s3.eu-central-1.amazonaws.com/container_registry/local.svg"
docs_url: Optional[str]
property
readonly
A URL to point at docs explaining this flavor.
Returns:
Type | Description |
---|---|
Optional[str] |
A flavor docs url. |
logo_url: str
property
readonly
A URL to represent the flavor in the dashboard.
Returns:
Type | Description |
---|---|
str |
The flavor logo. |
name: str
property
readonly
Name of the flavor.
Returns:
Type | Description |
---|---|
str |
The name of the flavor. |
sdk_docs_url: Optional[str]
property
readonly
A URL to point at docs explaining this flavor.
Returns:
Type | Description |
---|---|
Optional[str] |
A flavor docs url. |
dockerhub_container_registry
Implementation of a DockerHub Container Registry class.
DockerHubContainerRegistryFlavor (BaseContainerRegistryFlavor)
Class for DockerHub Container Registry.
Source code in zenml/container_registries/dockerhub_container_registry.py
class DockerHubContainerRegistryFlavor(BaseContainerRegistryFlavor):
"""Class for DockerHub Container Registry."""
@property
def name(self) -> str:
"""Name of the flavor.
Returns:
The name of the flavor.
"""
return ContainerRegistryFlavor.DOCKERHUB.value
@property
def service_connector_requirements(
self,
) -> Optional[ServiceConnectorRequirements]:
"""Service connector resource requirements for service connectors.
Specifies resource requirements that are used to filter the available
service connector types that are compatible with this flavor.
Returns:
Requirements for compatible service connectors, if a service
connector is required for this flavor.
"""
return ServiceConnectorRequirements(
connector_type="docker",
resource_type=DOCKER_REGISTRY_RESOURCE_TYPE,
resource_id_attr="uri",
)
@property
def docs_url(self) -> Optional[str]:
"""A url to point at docs explaining this flavor.
Returns:
A flavor docs url.
"""
return self.generate_default_docs_url()
@property
def sdk_docs_url(self) -> Optional[str]:
"""A url to point at docs explaining this flavor.
Returns:
A flavor docs url.
"""
return self.generate_default_sdk_docs_url()
@property
def logo_url(self) -> str:
"""A url to represent the flavor in the dashboard.
Returns:
The flavor logo.
"""
return "https://public-flavor-logos.s3.eu-central-1.amazonaws.com/container_registry/docker.png"
docs_url: Optional[str]
property
readonly
A url to point at docs explaining this flavor.
Returns:
Type | Description |
---|---|
Optional[str] |
A flavor docs url. |
logo_url: str
property
readonly
A url to represent the flavor in the dashboard.
Returns:
Type | Description |
---|---|
str |
The flavor logo. |
name: str
property
readonly
Name of the flavor.
Returns:
Type | Description |
---|---|
str |
The name of the flavor. |
sdk_docs_url: Optional[str]
property
readonly
A url to point at docs explaining this flavor.
Returns:
Type | Description |
---|---|
Optional[str] |
A flavor docs url. |
service_connector_requirements: Optional[zenml.models.v2.misc.service_connector_type.ServiceConnectorRequirements]
property
readonly
Service connector resource requirements for service connectors.
Specifies resource requirements that are used to filter the available service connector types that are compatible with this flavor.
Returns:
Type | Description |
---|---|
Optional[zenml.models.v2.misc.service_connector_type.ServiceConnectorRequirements] |
Requirements for compatible service connectors, if a service connector is required for this flavor. |
gcp_container_registry
Implementation of a GCP Container Registry class.
GCPContainerRegistryFlavor (BaseContainerRegistryFlavor)
Class for GCP Container Registry.
Source code in zenml/container_registries/gcp_container_registry.py
class GCPContainerRegistryFlavor(BaseContainerRegistryFlavor):
"""Class for GCP Container Registry."""
@property
def name(self) -> str:
"""Name of the flavor.
Returns:
The name of the flavor.
"""
return ContainerRegistryFlavor.GCP.value
@property
def service_connector_requirements(
self,
) -> Optional[ServiceConnectorRequirements]:
"""Service connector resource requirements for service connectors.
Specifies resource requirements that are used to filter the available
service connector types that are compatible with this flavor.
Returns:
Requirements for compatible service connectors, if a service
connector is required for this flavor.
"""
return ServiceConnectorRequirements(
connector_type="gcp",
resource_type=DOCKER_REGISTRY_RESOURCE_TYPE,
resource_id_attr="uri",
)
@property
def docs_url(self) -> Optional[str]:
"""A url to point at docs explaining this flavor.
Returns:
A flavor docs url.
"""
return self.generate_default_docs_url()
@property
def sdk_docs_url(self) -> Optional[str]:
"""A url to point at docs explaining this flavor.
Returns:
A flavor docs url.
"""
return self.generate_default_sdk_docs_url()
@property
def logo_url(self) -> str:
"""A url to represent the flavor in the dashboard.
Returns:
The flavor logo.
"""
return "https://public-flavor-logos.s3.eu-central-1.amazonaws.com/container_registry/gcp.png"
docs_url: Optional[str]
property
readonly
A url to point at docs explaining this flavor.
Returns:
Type | Description |
---|---|
Optional[str] |
A flavor docs url. |
logo_url: str
property
readonly
A url to represent the flavor in the dashboard.
Returns:
Type | Description |
---|---|
str |
The flavor logo. |
name: str
property
readonly
Name of the flavor.
Returns:
Type | Description |
---|---|
str |
The name of the flavor. |
sdk_docs_url: Optional[str]
property
readonly
A url to point at docs explaining this flavor.
Returns:
Type | Description |
---|---|
Optional[str] |
A flavor docs url. |
service_connector_requirements: Optional[zenml.models.v2.misc.service_connector_type.ServiceConnectorRequirements]
property
readonly
Service connector resource requirements for service connectors.
Specifies resource requirements that are used to filter the available service connector types that are compatible with this flavor.
Returns:
Type | Description |
---|---|
Optional[zenml.models.v2.misc.service_connector_type.ServiceConnectorRequirements] |
Requirements for compatible service connectors, if a service connector is required for this flavor. |
github_container_registry
Implementation of the GitHub Container Registry.
GitHubContainerRegistryConfig (BaseContainerRegistryConfig)
Configuration for the GitHub Container Registry.
Source code in zenml/container_registries/github_container_registry.py
class GitHubContainerRegistryConfig(BaseContainerRegistryConfig):
"""Configuration for the GitHub Container Registry."""
GitHubContainerRegistryFlavor (BaseContainerRegistryFlavor)
Class for GitHub Container Registry.
Source code in zenml/container_registries/github_container_registry.py
class GitHubContainerRegistryFlavor(BaseContainerRegistryFlavor):
"""Class for GitHub Container Registry."""
@property
def name(self) -> str:
"""Name of the flavor.
Returns:
The name of the flavor.
"""
return ContainerRegistryFlavor.GITHUB
@property
def docs_url(self) -> Optional[str]:
"""A url to point at docs explaining this flavor.
Returns:
A flavor docs url.
"""
return self.generate_default_docs_url()
@property
def sdk_docs_url(self) -> Optional[str]:
"""A url to point at docs explaining this flavor.
Returns:
A flavor docs url.
"""
return self.generate_default_sdk_docs_url()
@property
def logo_url(self) -> str:
"""A url to represent the flavor in the dashboard.
Returns:
The flavor logo.
"""
return "https://public-flavor-logos.s3.eu-central-1.amazonaws.com/container_registry/github.png"
docs_url: Optional[str]
property
readonly
A url to point at docs explaining this flavor.
Returns:
Type | Description |
---|---|
Optional[str] |
A flavor docs url. |
logo_url: str
property
readonly
A url to represent the flavor in the dashboard.
Returns:
Type | Description |
---|---|
str |
The flavor logo. |
name: str
property
readonly
Name of the flavor.
Returns:
Type | Description |
---|---|
str |
The name of the flavor. |
sdk_docs_url: Optional[str]
property
readonly
A url to point at docs explaining this flavor.
Returns:
Type | Description |
---|---|
Optional[str] |
A flavor docs url. |