Source code for runway.lookups.handlers.ecr

"""Retrieve a value from AWS Elastic Container Registry (ECR)."""
from __future__ import annotations

import base64
import logging
from typing import TYPE_CHECKING, Any, Union

from typing_extensions import Final, Literal

from ...lookups.handlers.base import LookupHandler

if TYPE_CHECKING:
    from mypy_boto3_ecr.client import ECRClient

    from ...context import CfnginContext, RunwayContext

LOGGER = logging.getLogger(__name__)


[docs]class EcrLookup(LookupHandler): """ECR Lookup.""" TYPE_NAME: Final[Literal["ecr"]] = "ecr" """Name that the Lookup is registered as."""
[docs] @staticmethod def get_login_password(client: ECRClient) -> str: """Get a password to login to ECR registry.""" auth = client.get_authorization_token().get("authorizationData", [None])[0] if not auth or "authorizationToken" not in auth: raise ValueError("get_authorization_token did not return authorizationData") auth_token = base64.b64decode(auth["authorizationToken"]).decode() _, password = auth_token.split(":") return password
[docs] @classmethod def handle( # pylint: disable=arguments-differ cls, value: str, context: Union[CfnginContext, RunwayContext], *__args: Any, **__kwargs: Any, ) -> Any: """Retrieve a value from AWS Elastic Container Registry (ECR). Args: value: The value passed to the Lookup. context: The current context object. """ query, args = cls.parse(value) session = context.get_session(region=args.get("region")) client = session.client("ecr") if query == "login-password": result = cls.get_login_password(client) else: raise ValueError(f"ecr lookup does not support '{query}'") return cls.format_results(result, **args)