Sono disponibili altri esempi AWS SDK nel repository AWS Doc SDK
Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.
Esempi di Amazon ECR che utilizzano SDK per Python (Boto3)
I seguenti esempi di codice mostrano come eseguire azioni e implementare scenari comuni utilizzando Amazon ECR. AWS SDK for Python (Boto3)
Le nozioni di base sono esempi di codice che mostrano come eseguire le operazioni essenziali all'interno di un servizio.
Le operazioni sono estratti di codice da programmi più grandi e devono essere eseguite nel contesto. Sebbene le operazioni mostrino come richiamare le singole funzioni del servizio, è possibile visualizzarle contestualizzate negli scenari correlati.
Ogni esempio include un collegamento al codice sorgente completo, dove puoi trovare istruzioni su come configurare ed eseguire il codice nel contesto.
Nozioni di base
I seguenti esempi di codice mostrano come iniziare a usare Amazon ECR.
- SDK per Python (Boto3)
-
Nota
C'è altro su. GitHub Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. import boto3 import argparse from boto3 import client def hello_ecr(ecr_client: client, repository_name: str) -> None: """ Use the AWS SDK for Python (Boto3) to create an Amazon Elastic Container Registry (Amazon ECR) client and list the images in a repository. This example uses the default settings specified in your shared credentials and config files. :param ecr_client: A Boto3 Amazon ECR Client object. This object wraps the low-level Amazon ECR service API. :param repository_name: The name of an Amazon ECR repository in your account. """ print( f"Hello, Amazon ECR! Let's list some images in the repository '{repository_name}':\n" ) paginator = ecr_client.get_paginator("list_images") page_iterator = paginator.paginate( repositoryName=repository_name, PaginationConfig={"MaxItems": 10} ) image_names: [str] = [] for page in page_iterator: for schedule in page["imageIds"]: image_names.append(schedule["imageTag"]) print(f"{len(image_names)} image(s) retrieved.") for schedule_name in image_names: print(f"\t{schedule_name}") if __name__ == "__main__": parser = argparse.ArgumentParser(description="Run hello Amazon ECR.") parser.add_argument( "--repository-name", type=str, help="the name of an Amazon ECR repository in your account.", required=True, ) args = parser.parse_args() hello_ecr(boto3.client("ecr"), args.repository_name)-
Per i dettagli sull'API, consulta ListImages in AWS SDK for Python (Boto3) API Reference.
-
Argomenti
Nozioni di base
L'esempio di codice seguente mostra come:
Crea un repository Amazon ECR.
Imposta le politiche del repository.
Recupera il repository. URIs
Ottieni i token di autorizzazione Amazon ECR.
Imposta le politiche del ciclo di vita per i repository Amazon ECR.
Invia un'immagine Docker a un repository Amazon ECR.
Verifica l'esistenza di un'immagine in un repository Amazon ECR.
Elenca i repository Amazon ECR per il tuo account e ottieni dettagli su di essi.
Elimina i repository Amazon ECR.
- SDK per Python (Boto3)
-
Nota
C'è altro da fare. GitHub Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. Esegui uno scenario interattivo al prompt dei comandi.
class ECRGettingStarted: """ A scenario that demonstrates how to use Boto3 to perform basic operations using Amazon ECR. """ def __init__( self, ecr_wrapper: ECRWrapper, docker_client: docker.DockerClient, ): self.ecr_wrapper = ecr_wrapper self.docker_client = docker_client self.tag = "echo-text" self.repository_name = "ecr-basics" self.docker_image = None self.full_tag_name = None self.repository = None def run(self, role_arn: str) -> None: """ Runs the scenario. """ print( """ The Amazon Elastic Container Registry (ECR) is a fully-managed Docker container registry service provided by AWS. It allows developers and organizations to securely store, manage, and deploy Docker container images. ECR provides a simple and scalable way to manage container images throughout their lifecycle, from building and testing to production deployment. The `ECRWrapper' class is a wrapper for the Boto3 'ecr' client. The 'ecr' client provides a set of methods to programmatically interact with the Amazon ECR service. This allows developers to automate the storage, retrieval, and management of container images as part of their application deployment pipelines. With ECR, teams can focus on building and deploying their applications without having to worry about the underlying infrastructure required to host and manage a container registry. This scenario walks you through how to perform key operations for this service. Let's get started... """ ) press_enter_to_continue() print_dashes() print( f""" * Create an ECR repository. An ECR repository is a private Docker container repository provided by Amazon Web Services (AWS). It is a managed service that makes it easy to store, manage, and deploy Docker container images. """ ) print(f"Creating a repository named {self.repository_name}") self.repository = self.ecr_wrapper.create_repository(self.repository_name) print(f"The ARN of the ECR repository is {self.repository['repositoryArn']}") repository_uri = self.repository["repositoryUri"] press_enter_to_continue() print_dashes() print( f""" * Build a Docker image. Create a local Docker image if it does not already exist. A Python Docker client is used to execute Docker commands. You must have Docker installed and running. """ ) print(f"Building a docker image from 'docker_files/Dockerfile'") self.full_tag_name = f"{repository_uri}:{self.tag}" self.docker_image = self.docker_client.images.build( path="docker_files", tag=self.full_tag_name )[0] print(f"Docker image {self.full_tag_name} successfully built.") press_enter_to_continue() print_dashes() if role_arn is None: print( """ * Because an IAM role ARN was not provided, a role policy will not be set for this repository. """ ) else: print( """ * Set an ECR repository policy. Setting an ECR repository policy using the `setRepositoryPolicy` function is crucial for maintaining the security and integrity of your container images. The repository policy allows you to define specific rules and restrictions for accessing and managing the images stored within your ECR repository. """ ) self.grant_role_download_access(role_arn) print(f"Download access granted to the IAM role ARN {role_arn}") press_enter_to_continue() print_dashes() print( """ * Display ECR repository policy. Now we will retrieve the ECR policy to ensure it was successfully set. """ ) policy_text = self.ecr_wrapper.get_repository_policy(self.repository_name) print("Policy Text:") print(f"{policy_text}") press_enter_to_continue() print_dashes() print( """ * Retrieve an ECR authorization token. You need an authorization token to securely access and interact with the Amazon ECR registry. The `get_authorization_token` method of the `ecr` client is responsible for securely accessing and interacting with an Amazon ECR repository. This operation is responsible for obtaining a valid authorization token, which is required to authenticate your requests to the ECR service. Without a valid authorization token, you would not be able to perform any operations on the ECR repository, such as pushing, pulling, or managing your Docker images. """ ) authorization_token = self.ecr_wrapper.get_authorization_token() print("Authorization token retrieved.") press_enter_to_continue() print_dashes() print( """ * Get the ECR Repository URI. The URI of an Amazon ECR repository is important. When you want to deploy a container image to a container orchestration platform like Amazon Elastic Kubernetes Service (EKS) or Amazon Elastic Container Service (ECS), you need to specify the full image URI, which includes the ECR repository URI. This allows the container runtime to pull the correct container image from the ECR repository. """ ) repository_descriptions = self.ecr_wrapper.describe_repositories( [self.repository_name] ) repository_uri = repository_descriptions[0]["repositoryUri"] print(f"Repository URI found: {repository_uri}") press_enter_to_continue() print_dashes() print( """ * Set an ECR Lifecycle Policy. An ECR Lifecycle Policy is used to manage the lifecycle of Docker images stored in your ECR repositories. These policies allow you to automatically remove old or unused Docker images from your repositories, freeing up storage space and reducing costs. This example policy helps to maintain the size and efficiency of the container registry by automatically removing older and potentially unused images, ensuring that the storage is optimized and the registry remains up-to-date. """ ) press_enter_to_continue() self.put_expiration_policy() print(f"An expiration policy was added to the repository.") print_dashes() print( """ * Push a docker image to the Amazon ECR Repository. The Docker client uses the authorization token is used to authenticate the when pushing the image to the ECR repository. """ ) decoded_authorization = base64.b64decode(authorization_token).decode("utf-8") username, password = decoded_authorization.split(":") resp = self.docker_client.api.push( repository=repository_uri, auth_config={"username": username, "password": password}, tag=self.tag, stream=True, decode=True, ) for line in resp: print(line) print_dashes() print("* Verify if the image is in the ECR Repository.") image_descriptions = self.ecr_wrapper.describe_images( self.repository_name, [self.tag] ) if len(image_descriptions) > 0: print("Image found in ECR Repository.") else: print("Image not found in ECR Repository.") press_enter_to_continue() print_dashes() print( "* As an optional step, you can interact with the image in Amazon ECR by using the CLI." ) if q.ask( "Would you like to view instructions on how to use the CLI to run the image? (y/n)", q.is_yesno, ): print( f""" 1. Authenticate with ECR - Before you can pull the image from Amazon ECR, you need to authenticate with the registry. You can do this using the AWS CLI: aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin {repository_uri.split("/")[0]} 2. Describe the image using this command: aws ecr describe-images --repository-name {self.repository_name} --image-ids imageTag={self.tag} 3. Run the Docker container and view the output using this command: docker run --rm {self.full_tag_name} """ ) self.cleanup(True) def cleanup(self, ask: bool): """ Deletes the resources created in this scenario. :param ask: If True, prompts the user to confirm before deleting the resources. """ if self.repository is not None and ( not ask or q.ask( f"Would you like to delete the ECR repository '{self.repository_name}? (y/n) " ) ): print(f"Deleting the ECR repository '{self.repository_name}'.") self.ecr_wrapper.delete_repository(self.repository_name) if self.full_tag_name is not None and ( not ask or q.ask( f"Would you like to delete the local Docker image '{self.full_tag_name}? (y/n) " ) ): print(f"Deleting the docker image '{self.full_tag_name}'.") self.docker_client.images.remove(self.full_tag_name) def grant_role_download_access(self, role_arn: str): """ Grants the specified role access to download images from the ECR repository. :param role_arn: The ARN of the role to grant access to. """ policy_json = { "Version": "2008-10-17", "Statement": [ { "Sid": "AllowDownload", "Effect": "Allow", "Principal": {"AWS": role_arn}, "Action": ["ecr:BatchGetImage"], } ], } self.ecr_wrapper.set_repository_policy( self.repository_name, json.dumps(policy_json) ) def put_expiration_policy(self): """ Puts an expiration policy on the ECR repository. """ policy_json = { "rules": [ { "rulePriority": 1, "description": "Expire images older than 14 days", "selection": { "tagStatus": "any", "countType": "sinceImagePushed", "countUnit": "days", "countNumber": 14, }, "action": {"type": "expire"}, } ] } self.ecr_wrapper.put_lifecycle_policy( self.repository_name, json.dumps(policy_json) ) if __name__ == "__main__": parser = argparse.ArgumentParser( description="Run Amazon ECR getting started scenario." ) parser.add_argument( "--iam-role-arn", type=str, default=None, help="an optional IAM role ARN that will be granted access to download images from a repository.", required=False, ) parser.add_argument( "--no-art", action="store_true", help="accessibility setting that suppresses art in the console output.", ) args = parser.parse_args() no_art = args.no_art iam_role_arn = args.iam_role_arn demo = None a_docker_client = None try: a_docker_client = docker.from_env() if not a_docker_client.ping(): raise docker.errors.DockerException("Docker is not running.") except docker.errors.DockerException as err: logging.error( """ The Python Docker client could not be created. Do you have Docker installed and running? Here is the error message: %s """, err, ) sys.exit("Error with Docker.") try: an_ecr_wrapper = ECRWrapper.from_client() demo = ECRGettingStarted(an_ecr_wrapper, a_docker_client) demo.run(iam_role_arn) except Exception as exception: logging.exception("Something went wrong with the demo!") if demo is not None: demo.cleanup(False)ECRWrapper classe che racchiude le azioni di Amazon ECR.
class ECRWrapper: def __init__(self, ecr_client: client): self.ecr_client = ecr_client @classmethod def from_client(cls) -> "ECRWrapper": """ Creates a ECRWrapper instance with a default Amazon ECR client. :return: An instance of ECRWrapper initialized with the default Amazon ECR client. """ ecr_client = boto3.client("ecr") return cls(ecr_client) def create_repository(self, repository_name: str) -> dict[str, any]: """ Creates an ECR repository. :param repository_name: The name of the repository to create. :return: A dictionary of the created repository. """ try: response = self.ecr_client.create_repository(repositoryName=repository_name) return response["repository"] except ClientError as err: if err.response["Error"]["Code"] == "RepositoryAlreadyExistsException": print(f"Repository {repository_name} already exists.") response = self.ecr_client.describe_repositories( repositoryNames=[repository_name] ) return self.describe_repositories([repository_name])[0] else: logger.error( "Error creating repository %s. Here's why %s", repository_name, err.response["Error"]["Message"], ) raise def delete_repository(self, repository_name: str): """ Deletes an ECR repository. :param repository_name: The name of the repository to delete. """ try: self.ecr_client.delete_repository( repositoryName=repository_name, force=True ) print(f"Deleted repository {repository_name}.") except ClientError as err: logger.error( "Couldn't delete repository %s.. Here's why %s", repository_name, err.response["Error"]["Message"], ) raise def set_repository_policy(self, repository_name: str, policy_text: str): """ Sets the policy for an ECR repository. :param repository_name: The name of the repository to set the policy for. :param policy_text: The policy text to set. """ try: self.ecr_client.set_repository_policy( repositoryName=repository_name, policyText=policy_text ) print(f"Set repository policy for repository {repository_name}.") except ClientError as err: if err.response["Error"]["Code"] == "RepositoryPolicyNotFoundException": logger.error("Repository does not exist. %s.", repository_name) raise else: logger.error( "Couldn't set repository policy for repository %s. Here's why %s", repository_name, err.response["Error"]["Message"], ) raise def get_repository_policy(self, repository_name: str) -> str: """ Gets the policy for an ECR repository. :param repository_name: The name of the repository to get the policy for. :return: The policy text. """ try: response = self.ecr_client.get_repository_policy( repositoryName=repository_name ) return response["policyText"] except ClientError as err: if err.response["Error"]["Code"] == "RepositoryPolicyNotFoundException": logger.error("Repository does not exist. %s.", repository_name) raise else: logger.error( "Couldn't get repository policy for repository %s. Here's why %s", repository_name, err.response["Error"]["Message"], ) raise def get_authorization_token(self) -> str: """ Gets an authorization token for an ECR repository. :return: The authorization token. """ try: response = self.ecr_client.get_authorization_token() return response["authorizationData"][0]["authorizationToken"] except ClientError as err: logger.error( "Couldn't get authorization token. Here's why %s", err.response["Error"]["Message"], ) raise def describe_repositories(self, repository_names: list[str]) -> list[dict]: """ Describes ECR repositories. :param repository_names: The names of the repositories to describe. :return: The list of repository descriptions. """ try: response = self.ecr_client.describe_repositories( repositoryNames=repository_names ) return response["repositories"] except ClientError as err: logger.error( "Couldn't describe repositories. Here's why %s", err.response["Error"]["Message"], ) raise def put_lifecycle_policy(self, repository_name: str, lifecycle_policy_text: str): """ Puts a lifecycle policy for an ECR repository. :param repository_name: The name of the repository to put the lifecycle policy for. :param lifecycle_policy_text: The lifecycle policy text to put. """ try: self.ecr_client.put_lifecycle_policy( repositoryName=repository_name, lifecyclePolicyText=lifecycle_policy_text, ) print(f"Put lifecycle policy for repository {repository_name}.") except ClientError as err: logger.error( "Couldn't put lifecycle policy for repository %s. Here's why %s", repository_name, err.response["Error"]["Message"], ) raise def describe_images( self, repository_name: str, image_ids: list[str] = None ) -> list[dict]: """ Describes ECR images. :param repository_name: The name of the repository to describe images for. :param image_ids: The optional IDs of images to describe. :return: The list of image descriptions. """ try: params = { "repositoryName": repository_name, } if image_ids is not None: params["imageIds"] = [{"imageTag": tag} for tag in image_ids] paginator = self.ecr_client.get_paginator("describe_images") image_descriptions = [] for page in paginator.paginate(**params): image_descriptions.extend(page["imageDetails"]) return image_descriptions except ClientError as err: logger.error( "Couldn't describe images. Here's why %s", err.response["Error"]["Message"], ) raise-
Per informazioni dettagliate sull'API, consulta i seguenti argomenti nella Documentazione di riferimento API SDK AWS per Python (Boto3).
-
Operazioni
Il seguente esempio di codice mostra come usare. CreateRepository
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. class ECRWrapper: def __init__(self, ecr_client: client): self.ecr_client = ecr_client @classmethod def from_client(cls) -> "ECRWrapper": """ Creates a ECRWrapper instance with a default Amazon ECR client. :return: An instance of ECRWrapper initialized with the default Amazon ECR client. """ ecr_client = boto3.client("ecr") return cls(ecr_client) def create_repository(self, repository_name: str) -> dict[str, any]: """ Creates an ECR repository. :param repository_name: The name of the repository to create. :return: A dictionary of the created repository. """ try: response = self.ecr_client.create_repository(repositoryName=repository_name) return response["repository"] except ClientError as err: if err.response["Error"]["Code"] == "RepositoryAlreadyExistsException": print(f"Repository {repository_name} already exists.") response = self.ecr_client.describe_repositories( repositoryNames=[repository_name] ) return self.describe_repositories([repository_name])[0] else: logger.error( "Error creating repository %s. Here's why %s", repository_name, err.response["Error"]["Message"], ) raise-
Per i dettagli sull'API, consulta CreateRepository AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. DeleteRepository
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. class ECRWrapper: def __init__(self, ecr_client: client): self.ecr_client = ecr_client @classmethod def from_client(cls) -> "ECRWrapper": """ Creates a ECRWrapper instance with a default Amazon ECR client. :return: An instance of ECRWrapper initialized with the default Amazon ECR client. """ ecr_client = boto3.client("ecr") return cls(ecr_client) def delete_repository(self, repository_name: str): """ Deletes an ECR repository. :param repository_name: The name of the repository to delete. """ try: self.ecr_client.delete_repository( repositoryName=repository_name, force=True ) print(f"Deleted repository {repository_name}.") except ClientError as err: logger.error( "Couldn't delete repository %s.. Here's why %s", repository_name, err.response["Error"]["Message"], ) raise-
Per i dettagli sull'API, consulta DeleteRepository AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. DescribeImages
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. class ECRWrapper: def __init__(self, ecr_client: client): self.ecr_client = ecr_client @classmethod def from_client(cls) -> "ECRWrapper": """ Creates a ECRWrapper instance with a default Amazon ECR client. :return: An instance of ECRWrapper initialized with the default Amazon ECR client. """ ecr_client = boto3.client("ecr") return cls(ecr_client) def describe_images( self, repository_name: str, image_ids: list[str] = None ) -> list[dict]: """ Describes ECR images. :param repository_name: The name of the repository to describe images for. :param image_ids: The optional IDs of images to describe. :return: The list of image descriptions. """ try: params = { "repositoryName": repository_name, } if image_ids is not None: params["imageIds"] = [{"imageTag": tag} for tag in image_ids] paginator = self.ecr_client.get_paginator("describe_images") image_descriptions = [] for page in paginator.paginate(**params): image_descriptions.extend(page["imageDetails"]) return image_descriptions except ClientError as err: logger.error( "Couldn't describe images. Here's why %s", err.response["Error"]["Message"], ) raise-
Per i dettagli sull'API, consulta DescribeImages AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. DescribeRepositories
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. class ECRWrapper: def __init__(self, ecr_client: client): self.ecr_client = ecr_client @classmethod def from_client(cls) -> "ECRWrapper": """ Creates a ECRWrapper instance with a default Amazon ECR client. :return: An instance of ECRWrapper initialized with the default Amazon ECR client. """ ecr_client = boto3.client("ecr") return cls(ecr_client) def describe_repositories(self, repository_names: list[str]) -> list[dict]: """ Describes ECR repositories. :param repository_names: The names of the repositories to describe. :return: The list of repository descriptions. """ try: response = self.ecr_client.describe_repositories( repositoryNames=repository_names ) return response["repositories"] except ClientError as err: logger.error( "Couldn't describe repositories. Here's why %s", err.response["Error"]["Message"], ) raise-
Per i dettagli sull'API, consulta DescribeRepositories AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. GetAuthorizationToken
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. class ECRWrapper: def __init__(self, ecr_client: client): self.ecr_client = ecr_client @classmethod def from_client(cls) -> "ECRWrapper": """ Creates a ECRWrapper instance with a default Amazon ECR client. :return: An instance of ECRWrapper initialized with the default Amazon ECR client. """ ecr_client = boto3.client("ecr") return cls(ecr_client) def get_authorization_token(self) -> str: """ Gets an authorization token for an ECR repository. :return: The authorization token. """ try: response = self.ecr_client.get_authorization_token() return response["authorizationData"][0]["authorizationToken"] except ClientError as err: logger.error( "Couldn't get authorization token. Here's why %s", err.response["Error"]["Message"], ) raise-
Per i dettagli sull'API, consulta GetAuthorizationToken AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. GetRepositoryPolicy
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. class ECRWrapper: def __init__(self, ecr_client: client): self.ecr_client = ecr_client @classmethod def from_client(cls) -> "ECRWrapper": """ Creates a ECRWrapper instance with a default Amazon ECR client. :return: An instance of ECRWrapper initialized with the default Amazon ECR client. """ ecr_client = boto3.client("ecr") return cls(ecr_client) def get_repository_policy(self, repository_name: str) -> str: """ Gets the policy for an ECR repository. :param repository_name: The name of the repository to get the policy for. :return: The policy text. """ try: response = self.ecr_client.get_repository_policy( repositoryName=repository_name ) return response["policyText"] except ClientError as err: if err.response["Error"]["Code"] == "RepositoryPolicyNotFoundException": logger.error("Repository does not exist. %s.", repository_name) raise else: logger.error( "Couldn't get repository policy for repository %s. Here's why %s", repository_name, err.response["Error"]["Message"], ) raise-
Per i dettagli sull'API, consulta GetRepositoryPolicy AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. PutLifeCyclePolicy
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. class ECRWrapper: def __init__(self, ecr_client: client): self.ecr_client = ecr_client @classmethod def from_client(cls) -> "ECRWrapper": """ Creates a ECRWrapper instance with a default Amazon ECR client. :return: An instance of ECRWrapper initialized with the default Amazon ECR client. """ ecr_client = boto3.client("ecr") return cls(ecr_client) def put_lifecycle_policy(self, repository_name: str, lifecycle_policy_text: str): """ Puts a lifecycle policy for an ECR repository. :param repository_name: The name of the repository to put the lifecycle policy for. :param lifecycle_policy_text: The lifecycle policy text to put. """ try: self.ecr_client.put_lifecycle_policy( repositoryName=repository_name, lifecyclePolicyText=lifecycle_policy_text, ) print(f"Put lifecycle policy for repository {repository_name}.") except ClientError as err: logger.error( "Couldn't put lifecycle policy for repository %s. Here's why %s", repository_name, err.response["Error"]["Message"], ) raiseEsempio che inserisce una politica sulla data di scadenza.
def put_expiration_policy(self): """ Puts an expiration policy on the ECR repository. """ policy_json = { "rules": [ { "rulePriority": 1, "description": "Expire images older than 14 days", "selection": { "tagStatus": "any", "countType": "sinceImagePushed", "countUnit": "days", "countNumber": 14, }, "action": {"type": "expire"}, } ] } self.ecr_wrapper.put_lifecycle_policy( self.repository_name, json.dumps(policy_json) )-
Per i dettagli sull'API, consulta PutLifeCyclePolicy AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. SetRepositoryPolicy
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. class ECRWrapper: def __init__(self, ecr_client: client): self.ecr_client = ecr_client @classmethod def from_client(cls) -> "ECRWrapper": """ Creates a ECRWrapper instance with a default Amazon ECR client. :return: An instance of ECRWrapper initialized with the default Amazon ECR client. """ ecr_client = boto3.client("ecr") return cls(ecr_client) def set_repository_policy(self, repository_name: str, policy_text: str): """ Sets the policy for an ECR repository. :param repository_name: The name of the repository to set the policy for. :param policy_text: The policy text to set. """ try: self.ecr_client.set_repository_policy( repositoryName=repository_name, policyText=policy_text ) print(f"Set repository policy for repository {repository_name}.") except ClientError as err: if err.response["Error"]["Code"] == "RepositoryPolicyNotFoundException": logger.error("Repository does not exist. %s.", repository_name) raise else: logger.error( "Couldn't set repository policy for repository %s. Here's why %s", repository_name, err.response["Error"]["Message"], ) raiseEsempio che concede l'accesso al download di un ruolo IAM.
def grant_role_download_access(self, role_arn: str): """ Grants the specified role access to download images from the ECR repository. :param role_arn: The ARN of the role to grant access to. """ policy_json = { "Version": "2008-10-17", "Statement": [ { "Sid": "AllowDownload", "Effect": "Allow", "Principal": {"AWS": role_arn}, "Action": ["ecr:BatchGetImage"], } ], } self.ecr_wrapper.set_repository_policy( self.repository_name, json.dumps(policy_json) )-
Per i dettagli sull'API, consulta SetRepositoryPolicy AWSSDK for Python (Boto3) API Reference.
-
