Core

Core ECR utilities.

---

ECRImage

ECRImage(
    *,
    account_id,
    region,
    repository_name,
    image_digest,
    image_manifest=None,
    client=None
)

Bases: ECRResourceBase

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def __init__(
    self,
    *,
    account_id: str,
    region: str,
    repository_name: str,
    image_digest: str,
    image_manifest: str | None = None,
    client: ECRClient | None = None,
) -> None:
    super().__init__(  # type: ignore[call-arg]
        account_id=account_id,  # pyright: ignore[reportCallIssue]
        region=region,  # pyright: ignore[reportCallIssue]
        repository_name=repository_name,  # pyright: ignore[reportCallIssue]
        image_digest=image_digest,  # pyright: ignore[reportCallIssue]
        client=client,  # type: ignore[arg-type]
    )
    self._image_manifest = image_manifest

add_image_tags

add_image_tags(*image_tags)

Add tags to image.

Parameters:

Name	Type	Description	Default
*image_tags	str	Tags to add to image.	()

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def add_image_tags(self, *image_tags: str):
    """Add tags to image.

    Args:
        *image_tags: Tags to add to image.
    """
    self.logger.info(f"Adding tags={image_tags} to {self.uri}")
    for tag in image_tags:
        self.put_image(image_tag=tag)

get_image_config

get_image_config()

Get ECR or docker image configuration json metadata.

Returns:

Type	Description
dict[str, Any]	Dictionary with image configuration.

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def get_image_config(self) -> dict[str, Any]:
    """Get ECR or docker image configuration json metadata.

    Returns:
        Dictionary with image configuration.
    """
    # get image_manifest (sha256: hash)
    config_digest = json.loads(self.image_manifest)["config"]["digest"]
    registry = ECRRegistry(account_id=self.account_id, region=self.region)
    ecr_login = registry.get_ecr_login()

    response = requests.get(
        url=f"https://{registry.uri}/v2/{self.repository_name}/blobs/{config_digest}",
        headers={"Authorization": f"Basic {ecr_login.auth_token}"},
    )
    response.raise_for_status()
    return response.json()

get_image_config_layer

get_image_config_layer()

Get the image config layer from image manifest.

The schema of the image manifest config layer is defined here: https://distribution.github.io/distribution/spec/manifest-v2-2/#image-manifest-field-descriptions

Note

While docker image manifests can have multiple formats, ECR only supports the schema defined in the link above, a v2 single image manifest. There is a manifest list, that describes multiple architectures, but ECR does not support this. This method assumes the image manifest is in the correct format.

Returns:

Type	Description
LayerTypeDef	Layer Type dict of the config object.

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def get_image_config_layer(self) -> LayerTypeDef:
    """Get the image config layer from image manifest.

    The schema of the image manifest config layer is defined here:
    https://distribution.github.io/distribution/spec/manifest-v2-2/#image-manifest-field-descriptions

    Note:
        While docker image manifests can have multiple formats, ECR only supports
        the schema defined in the link above, a v2 single image manifest. There is
        a manifest list, that describes multiple architectures, but ECR does not support
        this. This method assumes the image manifest is in the correct format.

    Returns:
        Layer Type dict of the config object.
    """
    image_manifest = json.loads(self.image_manifest)
    layer = image_manifest["config"]
    return LayerTypeDef(
        layerDigest=layer["digest"],
        layerAvailability="AVAILABLE",
        layerSize=layer["size"],
        mediaType=layer["mediaType"],
    )

get_image_detail

get_image_detail()

Get image detail of this image from ECR.

Returns:

Type	Description
ImageDetailTypeDef	Image detail dictionary.

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def get_image_detail(self) -> ImageDetailTypeDef:
    """Get image detail of this image from ECR.

    Returns:
        Image detail dictionary.
    """
    response = self.client.describe_images(
        repositoryName=self.repository_name,
        registryId=self.account_id,
        imageIds=[ImageIdentifierTypeDef(imageDigest=self.image_digest)],
    )
    image_details = response["imageDetails"]
    if len(image_details) == 0:
        raise ResourceNotFoundError(
            f"Could not resolve image detail for {self}"
        )  # pragma: no cover
    return image_details[0]

get_image_layers

get_image_layers()

Get layers from image manifest into ECR Layer objects.

The schema of the image manifest layers is defined here: https://distribution.github.io/distribution/spec/manifest-v2-2/#image-manifest-field-descriptions

Note

While docker image manifests can have multiple formats, ECR only supports the schema defined in the link above, a v2 single image manifest. There is a manifest list, that describes multiple architectures, but ECR does not support this. This method assumes the image manifest is in the correct format.

Returns:

Type	Description
list[LayerTypeDef]	List of ECR Image layers.

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def get_image_layers(self) -> list[LayerTypeDef]:
    """Get layers from image manifest into ECR Layer objects.

    The schema of the image manifest layers is defined here:
    https://distribution.github.io/distribution/spec/manifest-v2-2/#image-manifest-field-descriptions

    Note:
        While docker image manifests can have multiple formats, ECR only supports
        the schema defined in the link above, a v2 single image manifest. There is
        a manifest list, that describes multiple architectures, but ECR does not support
        this. This method assumes the image manifest is in the correct format.

    Returns:
        List of ECR Image layers.
    """
    image_manifest = json.loads(self.image_manifest)

    return [
        LayerTypeDef(
            layerDigest=layer["digest"],
            layerAvailability="AVAILABLE",
            layerSize=layer["size"],
            mediaType=layer["mediaType"],
        )
        for layer in image_manifest["layers"]
    ]

put_image

put_image(image_tag)

Make a call to put_image API to add image to ECR repository.

This method will add an image to the ECR repository. If the image already exists, it will not raise an error. Instead, it will log that the image already exists with the given tag.

Note

This operation does not push an image to the repository. It only adds the image manifest to the repository.

Parameters:

Name	Type	Description	Default
image_tag	str | None	Tag to associate with image. If None, image is untagged.	required

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def put_image(self, image_tag: str | None):
    """Make a call to put_image API to add image to ECR repository.

    This method will add an image to the ECR repository. If the image already exists,
    it will not raise an error. Instead, it will log that the image already
    exists with the given tag.

    Note:
        This operation does not push an image to the repository. It only adds
        the image manifest to the repository.

    Args:
        image_tag: Tag to associate with image. If None, image is untagged.
    """
    try:
        if image_tag is None:
            self.client.put_image(
                registryId=self.account_id,
                repositoryName=self.repository_name,
                imageManifest=self.image_manifest,
                imageDigest=self.image_digest,
            )
        else:
            self.client.put_image(
                registryId=self.account_id,
                repositoryName=self.repository_name,
                imageManifest=self.image_manifest,
                imageDigest=self.image_digest,
                imageTag=image_tag,
            )
    except ClientError as e:
        # IF we receive a ImageAlreadyExistsException,
        # then we have nothing to worry about. Otherwise, raise error.
        if get_client_error_code(e) != "ImageAlreadyExistsException":
            raise e
        self.logger.info(f"Image already exists with tag={image_tag}")
    else:
        self.logger.info(f"Added new image with tag={image_tag}")

ECRImageUri

Bases: ECRRepositoryUri

from_components classmethod

from_components(
    repository_name,
    image_tag=None,
    image_digest=None,
    account_id=None,
    region=None,
)

Generate an Image URI.

If account ID not provided, account Id of credentials is used. If region is not provided, region of credentials is used.

Parameters:

Name	Type	Description	Default
repository_name	str	Name of ECR repository.	required
image_tag	str | None	Tag associated with image. Defaults to None.	None
image_digest	str | None	The image digest. Defaults to None.	None
account_id	str | None	The registry ID. Defaults to None.	None
region	str | None	AWS region. Defaults to None.	None

Raises:

Type	Description
ValueError	If both or neither image tag / image digest are provided.

Returns:

Type	Description
ECRImageUri	ECR Image URI.

Source code in src/aibs_informatics_aws_utils/ecr/core.py

@classmethod
def from_components(  # type: ignore[override]
    cls,
    repository_name: str,
    image_tag: str | None = None,
    image_digest: str | None = None,
    account_id: str | None = None,
    region: str | None = None,
) -> "ECRImageUri":
    """Generate an Image URI.

    If account ID not provided, account Id of credentials is used.
    If region is not provided, region of credentials is used.

    Args:
        repository_name: Name of ECR repository.
        image_tag: Tag associated with image. Defaults to None.
        image_digest: The image digest. Defaults to None.
        account_id: The registry ID. Defaults to None.
        region: AWS region. Defaults to None.

    Raises:
        ValueError: If both or neither image tag / image digest are provided.

    Returns:
        ECR Image URI.
    """
    if (image_tag and image_digest) or (not image_tag and not image_digest):
        raise ValueError(
            "Must provide EITHER image tag OR image digest. "
            f"image_tag={image_tag}, image_digest={image_digest}"
        )
    repo_uri = ECRRepositoryUri.from_components(
        repository_name=repository_name, account_id=account_id, region=region
    )
    image_id = f"{'@' if image_digest else ':'}{image_digest or image_tag}"
    return ECRImageUri(f"{repo_uri}{image_id}")

ECRRegistry

ECRRegistry(client=None, **data)

Bases: ECRResource

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def __init__(self, client: ECRClient | None = None, **data: Any) -> None:
    super().__init__(**data)
    self._client = client

get_repositories

get_repositories(
    repository_name=None, repository_tags=None
)

Filter repositories based on resource tags specified.

Parameters:

Name	Type	Description	Default
repository_name	str | Pattern | None	Repository name or pattern to filter by.	None
repository_tags	list[ResourceTag] | None	List of resource tags to filter by.	None

Returns:

Type	Description
list[ECRRepository]	Filtered list of repositories with resource tags.

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def get_repositories(
    self,
    repository_name: str | re.Pattern | None = None,
    repository_tags: list[ResourceTag] | None = None,
) -> list[ECRRepository]:
    """Filter repositories based on resource tags specified.

    Args:
        repository_name: Repository name or pattern to filter by.
        repository_tags: List of resource tags to filter by.

    Returns:
        Filtered list of repositories with resource tags.
    """
    repositories = self.list_repositories()
    filtered_repos = []
    for repo in repositories:
        if repository_tags:
            repo_tags = repo.get_resource_tags()
            if not all([filter_tag in repo_tags for filter_tag in repository_tags]):
                continue
        if repository_name:
            if isinstance(repository_name, re.Pattern):
                if not repository_name.match(repo.repository_name):
                    continue
            elif repository_name not in repo.repository_name:
                continue
        filtered_repos.append(repo)
    return filtered_repos

list_repositories

list_repositories()

List all repositories in the Registry.

Returns:

Type	Description
list[ECRRepository]	List of repositories in the registry.

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def list_repositories(self) -> list[ECRRepository]:
    """List all repositories in the Registry.

    Returns:
        List of repositories in the registry.
    """

    paginator = self.client.get_paginator("describe_repositories")
    repositories: list[ECRRepository] = []
    for describe_repos_response in paginator.paginate(registryId=self.account_id):
        for repository in describe_repos_response["repositories"]:
            assert "repositoryName" in repository
            repositories.append(
                ECRRepository(
                    account_id=self.account_id,
                    region=self.region,
                    repository_name=repository["repositoryName"],
                )
            )
    return repositories

ECRRegistryUri

Bases: ValidatedStr

from_components classmethod

from_components(account_id=None, region=None, **kwargs)

Generate a Registry URI.

If account ID not provided, account Id of credentials is used. If region is not provided, region of credentials is used.

Parameters:

Name	Type	Description	Default
account_id	str | None	The registry ID. Defaults to None.	None
region	str | None	AWS region. Defaults to None.	None
**kwargs		Additional keyword arguments (unused).	{}

Returns:

Type	Description
ECRRegistryUri	AWS Registry URI.

Source code in src/aibs_informatics_aws_utils/ecr/core.py

@classmethod
def from_components(
    cls, account_id: str | None = None, region: str | None = None, **kwargs
) -> "ECRRegistryUri":
    """Generate a Registry URI.

    If account ID not provided, account Id of credentials is used.
    If region is not provided, region of credentials is used.

    Args:
        account_id: The registry ID. Defaults to None.
        region: AWS region. Defaults to None.
        **kwargs: Additional keyword arguments (unused).

    Returns:
        AWS Registry URI.
    """
    account_id = account_id or get_account_id()
    region = get_region(region)
    return ECRRegistryUri(f"{account_id}.dkr.ecr.{region}.amazonaws.com")

ECRRepository

ECRRepository(client=None, **data)

Bases: ECRResource

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def __init__(self, client: ECRClient | None = None, **data: Any) -> None:
    super().__init__(**data)
    self._client = client

create

create(
    tags=None,
    image_tag_mutability="MUTABLE",
    exists_ok=True,
)

Create an ECR Repository.

Parameters:

Name	Type	Description	Default
tags	list[ResourceTag] | None	List of repo tags to add. Defaults to None.	None
image_tag_mutability	ImageTagMutabilityType	Whether image tag is immutable. Defaults to "MUTABLE".	'MUTABLE'
exists_ok	bool	Suppress error if repository already exists. Defaults to True.	True

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def create(
    self,
    tags: list[ResourceTag] | None = None,
    image_tag_mutability: ImageTagMutabilityType = "MUTABLE",
    exists_ok: bool = True,
):
    """Create an ECR Repository.

    Args:
        tags: List of repo tags to add. Defaults to None.
        image_tag_mutability: Whether image tag is immutable. Defaults to "MUTABLE".
        exists_ok: Suppress error if repository already exists. Defaults to True.
    """
    if tags is None:
        tags = []

    try:
        self.client.create_repository(
            registryId=self.account_id,
            repositoryName=self.repository_name,
            tags=tags,
            imageTagMutability=image_tag_mutability,
        )
    except ClientError as e:
        # If the repo already exists, just move on.
        if get_client_error_code(e) == "RepositoryAlreadyExistsException" and exists_ok:
            # update tags
            if tags:
                self.update_resource_tags(*tags)
        else:
            raise e

delete

delete(force)

Delete the ECR repository described by this instance.

Parameters:

Name	Type	Description	Default
force	bool	Ignore if images in repository.	required

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def delete(self, force: bool):
    """Delete the ECR repository described by this instance.

    Args:
        force: Ignore if images in repository.
    """
    self.client.delete_repository(
        registryId=self.account_id, repositoryName=self.repository_name, force=force
    )

exists

exists()

Check if repository exists.

Returns:

Type	Description
bool	True if repository exists.

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def exists(self) -> bool:
    """Check if repository exists.

    Returns:
        True if repository exists.
    """
    try:
        self.client.describe_repositories(
            registryId=self.account_id, repositoryNames=[self.repository_name]
        )
    except ClientError as e:
        if get_client_error_code(e) == "RepositoryNotFoundException":
            return False
        else:
            raise e
    else:
        return True

get_image

get_image(image_tag=None, image_digest=None)

Get the image associated with the following tag or digest.

Parameters:

Name	Type	Description	Default
image_tag	str | None	Image tag. Defaults to None.	None
image_digest	str | None	Image digest. Defaults to None.	None

Returns:

Type	Description
ECRImage	The image with the image tag or digest.

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def get_image(
    self, image_tag: str | None = None, image_digest: str | None = None
) -> "ECRImage":
    """Get the image associated with the following tag or digest.

    Args:
        image_tag: Image tag. Defaults to None.
        image_digest: Image digest. Defaults to None.

    Returns:
        The image with the image tag or digest.
    """
    if (image_tag is None) == (image_digest is None):
        raise ValueError(
            f"Must provide image tag XOR digest. "
            f"provided image_tag={image_tag}, image_digest={image_digest}"
        )
    images = self.get_images()
    for image in images:
        if image_tag and image_tag in image.image_tags:
            return image
        elif image_digest and image_digest == image.image_digest:
            return image
    else:
        raise ResourceNotFoundError(
            f"Could not find an image in {self.uri} with tag={image_tag}"
        )

get_images

get_images(tag_status='ANY')

Fetches all images in a given repository.

Parameters:

Name	Type	Description	Default
tag_status	TagStatusType	Filter non-tagged images. Defaults to "ANY".	'ANY'

Returns:

Type	Description
list[ECRImage]	List of images in repository.

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def get_images(self, tag_status: TagStatusType = "ANY") -> list["ECRImage"]:
    """Fetches all images in a given repository.

    Args:
        tag_status: Filter non-tagged images. Defaults to "ANY".

    Returns:
        List of images in repository.
    """

    # To fetch necessary image info, we need to make two calls:
    #   1. Call the `list_images` API which will return imageDigest/imageTag.
    #       - This is a lightweight response which gives us digests
    #   2. Call the `batch_get_image` API which returns imageManifest info.
    #       - this is consolidated all in one go

    # Call 1: list_images
    paginator = self.client.get_paginator("list_images")
    list_image_request = ListImagesRequestPaginateTypeDef(
        repositoryName=self.repository_name,
        filter=ListImagesFilterTypeDef(tagStatus=tag_status),
    )
    image_digests: list[str] = sorted(
        list(
            {
                image_id["imageDigest"]
                for list_images_response in paginator.paginate(**list_image_request)
                for image_id in list_images_response["imageIds"]
                if "imageDigest" in image_id
            }
        )
    )
    if len(image_digests) == 0:
        return []

    # Call 2: batch_get_image
    response: BatchGetImageResponseTypeDef = self.client.batch_get_image(
        repositoryName=self.repository_name,
        registryId=self.account_id,
        imageIds=[ImageIdentifierTypeDef(imageDigest=digest) for digest in image_digests],
    )

    # Next we consolidate the results, ensuring that the image manifests
    # are all the same. If an image digest has differing manifests,
    # we should throw an error.
    digest_to_manifest_map: dict[str, str] = {}
    for image in response["images"]:
        image_digest = image["imageId"]["imageDigest"]  # type: ignore
        image_manifest = image["imageManifest"]  # type: ignore
        if image_digest in digest_to_manifest_map:
            if image_manifest != digest_to_manifest_map[image_digest]:
                raise ValueError(
                    f"Not all image manifests are equivalent for {image_digest} in {self.uri}"
                )
        else:
            digest_to_manifest_map[image_digest] = image_manifest

    return [
        ECRImage(
            account_id=self.account_id,
            region=self.region,
            repository_name=self.repository_name,
            image_digest=image_digest,
            image_manifest=image_manifest,
        )
        for image_digest, image_manifest in digest_to_manifest_map.items()
    ]

ECRRepositoryUri

Bases: ECRRegistryUri

from_components classmethod

from_components(
    repository_name, account_id=None, region=None
)

Generate a Repository URI.

If account ID not provided, account Id of credentials is used. If region is not provided, region of credentials is used.

Parameters:

Name	Type	Description	Default
repository_name	str	Name of ECR repository.	required
account_id	str | None	The registry ID. Defaults to None.	None
region	str | None	AWS region. Defaults to None.	None

Returns:

Type	Description
ECRRepositoryUri	AWS Repository URI.

Source code in src/aibs_informatics_aws_utils/ecr/core.py

@classmethod
def from_components(  # type: ignore[override]
    cls,
    repository_name: str,
    account_id: str | None = None,
    region: str | None = None,
) -> "ECRRepositoryUri":
    """Generate a Repository URI.

    If account ID not provided, account Id of credentials is used.
    If region is not provided, region of credentials is used.

    Args:
        repository_name: Name of ECR repository.
        account_id: The registry ID. Defaults to None.
        region: AWS region. Defaults to None.

    Returns:
        AWS Repository URI.
    """
    registry_uri = ECRRegistryUri.from_components(account_id=account_id, region=region)
    return ECRRepositoryUri(f"{registry_uri}/{repository_name}")

ECRResource

ECRResource(client=None, **data)

Bases: ECRResourceBase

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def __init__(self, client: ECRClient | None = None, **data: Any) -> None:
    super().__init__(**data)
    self._client = client

get_resource_tags

get_resource_tags()

Gets the tags for this ECR Resource.

Returns:

Type	Description
list[ResourceTag]	List of tags.

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def get_resource_tags(self) -> list[ResourceTag]:
    """Gets the tags for this ECR Resource.

    Returns:
        List of tags.
    """
    return [
        ResourceTag(Key=tag["Key"], Value=tag["Value"])
        for tag in self.client.list_tags_for_resource(resourceArn=self.arn)["tags"]
        if "Key" in tag and "Value" in tag
    ]

update_resource_tags

update_resource_tags(
    *tags, mode=cast(TagMode, TagMode.APPEND)
)

Updates the tags for an ECR Resource.

An update can either append or overwrite the existing tags.

Parameters:

Name	Type	Description	Default
*tags	ResourceTag	Resource tags to update.	()
mode	TagMode	Either append or overwrite tags of resource. Defaults to TagMode.APPEND.	cast(TagMode, APPEND)

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def update_resource_tags(
    self, *tags: ResourceTag, mode: TagMode = cast(TagMode, TagMode.APPEND)
):
    """Updates the tags for an ECR Resource.

    An update can either append or overwrite the existing tags.

    Args:
        *tags: Resource tags to update.
        mode: Either append or overwrite tags of resource.
            Defaults to TagMode.APPEND.
    """
    tag_dict = {tag["Key"]: tag["Value"] for tag in tags}
    if mode == TagMode.OVERWRITE:
        existing_tags = self.get_resource_tags()
        tag_keys_to_remove = [
            existing_tag["Key"]
            for existing_tag in existing_tags
            if existing_tag["Key"] not in tag_dict
        ]
        self.client.untag_resource(resourceArn=self.arn, tagKeys=tag_keys_to_remove)
    self.client.tag_resource(
        resourceArn=self.arn,
        tags=[TagTypeDef(Key=key, Value=value) for key, value in tag_dict.items()],
    )

resolve_image_uri

resolve_image_uri(name, default_tag=None)

Resolve full image URI from input name.

Parameters:

Name	Type	Description	Default
name	str	Partial or fully qualified uri, name of image or repository.	required
default_tag	str | None	Default tag to use if not specified. Defaults to None.	None

Returns:

Type	Description
str	Fully qualified image URI.

Source code in src/aibs_informatics_aws_utils/ecr/core.py

def resolve_image_uri(name: str, default_tag: str | None = None) -> str:
    """Resolve full image URI from input name.

    Args:
        name: Partial or fully qualified uri, name of image or repository.
        default_tag: Default tag to use if not specified. Defaults to None.

    Returns:
        Fully qualified image URI.
    """

    try:
        uri = name

        if not ECRRegistryUri.is_prefixed(uri):
            uri = f"{ECRRegistryUri.from_components()}/{uri}"

        if ECRImageUri.is_valid(uri):
            return uri
        elif ECRRepositoryUri.is_valid(uri):
            repo = ECRRepository.from_uri(uri)

            if default_tag:
                image = repo.get_image(image_tag=default_tag)
                return image.uri
            else:
                # Fetch latest tagged
                def get_image_push_time(image: ECRImage) -> datetime:
                    if image.image_pushed_at is None:
                        raise RuntimeError(f"Couldn't get 'image_pushed_at' for: {image}")
                    return image.image_pushed_at

                images = sorted(repo.get_images("TAGGED"), key=get_image_push_time)
                return images[-1].uri
        else:
            raise ValueError(f"Could not resolve full URI for image {uri} (raw={name})")
    except Exception as e:
        msg = f"Couldn't resolve ECR image URI from {name} with error: {e}"
        logger.exception(msg)
        raise ResourceNotFoundError(msg) from e