ECR¶

Utilities for working with Amazon Elastic Container Registry.

Overview¶

Module	Description
Core	Core ECR utilities
Image Replicator	Container image replication

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}")

ECRImageReplicator ¶

Bases: LoggingMixin

put_image ¶

put_image(
    source_image,
    destination_repository,
    destination_image_tags=None,
)

Put image manifest and tags for an image.

This is the final step in copying an ECR image. It must be done once all the layers of the image have been uploaded.

If the put_image request fails with a 'LayersNotFoundException', this will attempt to upload the missing layers and retry the put_image request.

Parameters:

Name	Type	Description	Default
`source_image`	`ECRImage`	Source image that has been copied.	required
`destination_repository`	`ECRRepository`	Destination repository.	required
`destination_image_tags`	`list[str] \| None`	Destination image tags. Optional. If not provided, source image's tags are added.	`None`

Returns:

Type	Description
`ECRImage`	Destination ECR Image.

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

def put_image(
    self,
    source_image: ECRImage,
    destination_repository: ECRRepository,
    destination_image_tags: list[str] | None = None,
) -> ECRImage:
    """Put image manifest and tags for an image.

    This is the final step in copying an ECR image. It must be done
    once all the layers of the image have been uploaded.

    If the put_image request fails with a 'LayersNotFoundException',
    this will attempt to upload the missing layers and retry
    the put_image request.

    Args:
        source_image: Source image that has been copied.
        destination_repository: Destination repository.
        destination_image_tags: Destination image tags. Optional.
            If not provided, source image's tags are added.

    Returns:
        Destination ECR Image.
    """
    tags = (
        source_image.image_tags if destination_image_tags is None else destination_image_tags
    )

    self.logger.info(f"putting image to {destination_repository.uri} with {tags} tags")
    dest_image = ECRImage(
        account_id=destination_repository.account_id,
        region=destination_repository.region,
        repository_name=destination_repository.repository_name,
        image_digest=source_image.image_digest,
        image_manifest=source_image.image_manifest,
        client=destination_repository.client,
    )
    try:
        dest_image.put_image(None)
        if tags:
            dest_image.add_image_tags(*tags)

    except ClientError as e:
        # IF we have a LayersNotFoundException, then we will retry to upload layers
        if get_client_error_code(e) != "LayersNotFoundException":
            self.log_stacktrace(
                f"Error putting image to {destination_repository.uri} with {tags}",
                e,
            )
            raise e
        self.logger.warning(
            f"Received 'LayerNotFoundException' while putting image. "
            f"Resolving missing layers of {source_image} specified in {e}"
        )
        missing_layers = self._get_missing_layers(
            client=source_image.client,
            repository_name=source_image.repository_name,
            put_image_error=e,
        )
        self.logger.info(f"Identified {len(missing_layers)} missing layers. Uploading layers")
        self._upload_layers(
            source_repository=source_image.get_repository(),
            destination_repository=destination_repository,
            layers=missing_layers,
            check_if_exists=False,
        )

        self.logger.info("uploaded all missing layers, retrying put_image request")
        self.put_image(source_image, destination_repository, destination_image_tags)
    return dest_image

replicate ¶

replicate(
    source_image,
    destination_repository,
    destination_image_tags=None,
)

Copies the source image into the destination repository.

This allows the user to facilitate replication:

between AWS accounts
to a new repository
with new tags

Note

This has exactly the same effect as docker pull; docker tag; docker push, but is done with the AWS ECR SDK so we can run this in a lambda when our stack updates.

Parameters:

Name	Type	Description	Default
`source_image`	`ECRImage`	Configuration of image to pull.	required
`destination_repository`	`ECRRepository`	Repository to push to.	required
`destination_image_tags`	`list[str] \| None`	Optional list of tags to apply to the destination image.	`None`

Returns:

Type	Description
`ECRImage`	Configuration of destination image pushed.

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

def replicate(
    self,
    source_image: ECRImage,
    destination_repository: ECRRepository,
    destination_image_tags: list[str] | None = None,
) -> ECRImage:
    """Copies the source image into the destination repository.

    This allows the user to facilitate replication:

    - **between AWS accounts**
    - **to a new repository**
    - **with new tags**

    Note:
        This has exactly the same effect as `docker pull; docker tag; docker push`,
        but is done with the AWS ECR SDK so we can run this in a lambda when our
        stack updates.

    Args:
        source_image: Configuration of image to pull.
        destination_repository: Repository to push to.
        destination_image_tags: Optional list of tags to apply to the destination image.

    Returns:
        Configuration of destination image pushed.
    """
    self.log.info(
        f"Starting Image Replication "
        f"[source={source_image.uri}, destination={destination_repository.uri}] "
        f"with tags={destination_image_tags}"
    )

    self.upload_layers(
        source_image=source_image, destination_repository=destination_repository
    )

    self.logger.info(f"Putting image to {destination_repository.uri}")

    self.put_image(
        source_image=source_image,
        destination_repository=destination_repository,
        destination_image_tags=destination_image_tags,
    )
    self.logger.info(f"Completed putting image to {destination_repository.uri}.")

    return ECRImage(
        account_id=destination_repository.account_id,
        region=destination_repository.region,
        repository_name=destination_repository.repository_name,
        image_digest=source_image.image_digest,
        client=destination_repository.client,
    )

upload_layers ¶

upload_layers(source_image, destination_repository)

Upload image layers of the source image to the destination repository.

Parameters:

Name	Type	Description	Default
`source_image`	`ECRImage`	Source image that has been copied.	required
`destination_repository`	`ECRRepository`	Destination repository.	required

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

def upload_layers(self, source_image: ECRImage, destination_repository: ECRRepository):
    """Upload image layers of the source image to the destination repository.

    Args:
        source_image: Source image that has been copied.
        destination_repository: Destination repository.
    """
    self.logger.info(
        f"Uploading layers from {source_image.uri} to {destination_repository.uri}."
    )

    layers = source_image.get_image_layers()
    config_layer = source_image.get_image_config_layer()

    all_layers = layers + [config_layer]

    self.logger.info(
        f"Source image {source_image.uri} has {len(layers)} layers and "
        f"1 config layer totalling {len(all_layers)} layers"
    )

    self._upload_layers(
        source_repository=source_image.get_repository(),
        destination_repository=destination_repository,
        layers=all_layers,
        check_if_exists=True,
    )

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

ECR¶

Overview¶

ECRImage ¶

add_image_tags ¶

get_image_config ¶

get_image_config_layer ¶

get_image_detail ¶

get_image_layers ¶

put_image ¶

ECRImageReplicator ¶

put_image ¶

replicate ¶

upload_layers ¶

ECRImageUri ¶

from_components classmethod ¶

ECRRegistry ¶

get_repositories ¶

list_repositories ¶

ECRRegistryUri ¶

from_components classmethod ¶

ECRRepository ¶

create ¶

delete ¶

exists ¶

get_image ¶

get_images ¶

ECRRepositoryUri ¶

from_components classmethod ¶

ECRResource ¶

get_resource_tags ¶

update_resource_tags ¶

resolve_image_uri ¶

from_components `classmethod` ¶

from_components `classmethod` ¶

from_components `classmethod` ¶