Source code for runway.cfngin.hooks.docker.image._remove

"""Docker image remove hook.

Replicates the functionality of the ``docker image remove`` CLI command.

# pylint: disable=no-self-argument
from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any, Dict, List, Optional

from docker.errors import ImageNotFound
from pydantic import Field, validator

from .....utils import BaseModel
from ..data_models import (
from ..hook_data import DockerHookData

    from .....context import CfnginContext

LOGGER = logging.getLogger(__name__.replace("._", "."))

[docs]class ImageRemoveArgs(BaseModel): """Args passed to image.remove.""" _ctx: Optional[CfnginContext] = Field(default=None, alias="context", export=False) ecr_repo: Optional[ElasticContainerRegistryRepository] = None # depends on _ctx """AWS Elastic Container Registry repository information. Providing this will automatically construct the repo URI. If provided, do not provide ``repo``. If using a private registry, only ``repo_name`` is required. If using a public registry, ``repo_name`` and ``registry_alias``. """ force: bool = False """Whether to force the removal of the image.""" image: Optional[DockerImage] = None """Image to push.""" noprune: bool = False """Whether to delete untagged parents.""" repo: Optional[str] = None # depends on ecr_repo & image """URI of a non Docker Hub repository where the image will be stored.""" tags: List[str] = [] # depends on image """List of tags to remove.""" @validator("ecr_repo", pre=True, allow_reuse=True) def _set_ecr_repo(cls, v: Any, values: Dict[str, Any]) -> Any: """Set the value of ``ecr_repo``.""" if v and isinstance(v, dict): return ElasticContainerRegistryRepository.parse_obj( { "repo_name": v.get("repo_name"), "registry": ElasticContainerRegistry.parse_obj( { "account_id": v.get("account_id"), "alias": v.get("registry_alias"), "aws_region": v.get("aws_region"), "context": values.get("context"), } ), } ) return v @validator("repo", pre=True, always=True, allow_reuse=True) def _set_repo(cls, v: Optional[str], values: Dict[str, Any]) -> Optional[str]: """Set the value of ``repo``.""" if v: return v image: Optional[DockerImage] = values.get("image") if image: return image.repo ecr_repo: Optional[ElasticContainerRegistryRepository] = values.get("ecr_repo") if ecr_repo: return ecr_repo.fqn return None @validator("tags", pre=True, always=True, allow_reuse=True) def _set_tags(cls, v: List[str], values: Dict[str, Any]) -> List[str]: """Set the value of ``tags``.""" if v: return v image: Optional[DockerImage] = values.get("image") if image: return image.tags return ["latest"]
[docs]def remove(*, context: CfnginContext, **kwargs: Any) -> DockerHookData: """Docker image push remove. Replicates the functionality of ``docker image push`` CLI command. kwargs are parsed by :class:`~runway.cfngin.hooks.docker.image.ImageRemoveArgs`. """ args = ImageRemoveArgs.parse_obj({"context": context, **kwargs}) docker_hook_data = DockerHookData.from_cfngin_context(context)"removing local image %s...", args.repo) for tag in args.tags: image = f"{args.repo}:{tag}" try: docker_hook_data.client.images.remove( image=image, force=args.force, noprune=args.noprune )"successfully removed local image %s", image) except ImageNotFound: LOGGER.warning("local image %s does not exist", image) if docker_hook_data.image and kwargs.get("image"): if kwargs["image"].id == docker_hook_data.image = None # clear out the image that was set return docker_hook_data.update_context(context)