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, None otherwise.

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	True if the container registry requires authentication, False otherwise.

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
    default_repository: Optional[str] = None

    @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.