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.service_connector_models.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.service_connector_models.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_authentication_secret(
            expected_schema_type=BasicAuthSecretSchema
        )
        if secret:
            return secret.username, secret.password

        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 = DockerClient.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) pydantic-model

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

    @validator("uri")
    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.

        This designation is used to determine if the stack component can be
        shared with other users or if it is only usable on the local host.

        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.

This designation is used to determine if the stack component can be shared with other users or if it is only usable on the local host.

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

@validator("uri")
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.service_connector_models.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.service_connector_models.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.service_connector_models.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.service_connector_models.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.service_connector_models.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.service_connector_models.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) pydantic-model

Configuration for the GitHub Container Registry.

Attributes:

Name	Type	Description
automatic_token_authentication	bool	If True, use automatic token authentication (https://docs.github.com/en/actions/security-guides/automatic-token-authentication#using-the-github_token-in-a-workflow) when trying to access this container registry from within a GitHub Actions environment.

Source code in zenml/container_registries/github_container_registry.py

class GitHubContainerRegistryConfig(BaseContainerRegistryConfig):
    """Configuration for the GitHub Container Registry.

    Attributes:
        automatic_token_authentication: If `True`, use automatic token
            authentication (https://docs.github.com/en/actions/security-guides/automatic-token-authentication#using-the-github_token-in-a-workflow)
            when trying to access this container registry from within a GitHub
            Actions environment.
    """

    automatic_token_authentication: bool = False

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.