"""Docker image push hook.
Replicates the functionality of the ``docker image push`` CLI command.
.. rubric:: Hook Path
``runway.cfngin.hooks.docker.image.push``
.. rubric:: Args
ecr_repo (Optional[Dict[str, Optional[str]]])
Information describing an ECR repository. This is used to construct the repository URL.
If providing a value for this field, do not provide a value for ``image`` or ``repo``.
If using a private registry, only ``repo_name`` is required.
If using a public registry, ``repo_name`` and ``registry_alias``.
account_id (Optional[str])
AWS account ID that owns the registry being logged into. If not provided,
it will be acquired automatically if needed.
aws_region (Optional[str])
AWS region where the registry is located. If not provided, it will be acquired
automatically if needed.
registry_alias (Optional[str])
If it is a public repository, provide the alias.
repo_name (str)
The name of the repository
image (Optional[DockerImage])
A :class:`~runway.cfngin.hooks.docker.data_models.DockerImage` object.
This can be retrieved from ``hook_data`` for a preceding *build* using the
:ref:`hook_data Lookup <hook_data lookup>`.
If providing a value for this field, do not provide a value for ``ecr_repo`` or ``repo``.
repo (Optional[str])
URI of a non Docker Hub repository where the image will be stored.
If providing one of the other repo values or ``image``, leave this value empty.
tags (Optional[List[str]])
List of tags push. (*default:* ``["latest"]``)
.. rubric:: Example
.. code-block:: yaml
pre_build:
- path: runway.cfngin.hooks.docker.login
args:
ecr: true
password: ${ecr login-password}
- path: runway.cfngin.hooks.docker.image.build
args:
ecr_repo:
repo_name: ${cfn ${namespace}-test-ecr.Repository}
tags:
- latest
- python3.9
- path: runway.cfngin.hooks.docker.image.push
args:
image: ${hook_data docker.image}
stacks:
ecr-lambda-function:
class_path: blueprints.EcrFunction
variables:
ImageUri: ${hook_data docker.image.uri.latest}
"""
import logging
from typing import TYPE_CHECKING, Any, Dict, List, Optional
from ..data_models import BaseModel, DockerImage, ElasticContainerRegistryRepository
from ..hook_data import DockerHookData
if TYPE_CHECKING:
from ....context import Context
LOGGER = logging.getLogger(__name__.replace("._", "."))
class ImagePushArgs(BaseModel):
"""Args passed to image.push."""
def __init__(
self,
ecr_repo=None, # type: Optional[Dict[str, Any]]
image=None, # type: Optional[DockerImage]
repo=None, # type: Optional[str]
tags=None, # type: Optional[List[str]]
**kwargs # type: Any
): # type: (...) -> None
"""Instantiate class."""
self.repo = self.determine_repo(
context=kwargs.get("context"), ecr_repo=ecr_repo, image=image, repo=repo
)
if image and not tags:
tags = image.tags
self.tags = self._validate_list_str(tags or ["latest"], required=True)
@staticmethod
def determine_repo(
context=None, # type: Optional["Context"]
ecr_repo=None, # type: Optional[Dict[str, Optional[str]]]
image=None, # type: Optional[DockerImage]
repo=None, # type: Optional[str]
): # type: (...) -> Optional[str]
"""Determine repo URI.
Args:
context: CFNgin context.
ecr_repo: AWS Elastic Container Registry options.
image: Docker image object.
repo: URI of a non Docker Hub repository.
"""
if repo:
return repo
if isinstance(image, DockerImage):
return image.repo
if ecr_repo:
return ElasticContainerRegistryRepository.parse_obj(
ecr_repo, context=context
).fqn
return None
[docs]def push(**kwargs): # type: (...) -> DockerHookData
"""Docker image push hook.
Replicates the functionality of ``docker image push`` CLI command.
"""
context = kwargs.pop("context") # type: "Context"
kwargs.pop("provider", None) # not needed
args = ImagePushArgs.parse_obj(kwargs, context=context)
docker_hook_data = DockerHookData.from_cfngin_context(context)
LOGGER.info("pushing image %s...", args.repo)
for tag in args.tags:
docker_hook_data.client.images.push(repository=args.repo, tag=tag)
LOGGER.info("successfully pushed image %s:%s", args.repo, tag)
return docker_hook_data.update_context(context)