Weitere AWS SDK Beispiele sind im Repo AWS Doc SDK Examples
Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.
Archivieren Sie eine Datei in Amazon S3 Glacier, erhalten Sie Benachrichtigungen und initiieren Sie einen Job mit einem AWS SDK
Wie das aussehen kann, sehen Sie am nachfolgenden Beispielcode:
Erstellen eines Amazon-S3-Glacier-Tresors.
Konfigurieren Sie den Tresor so, dass Benachrichtigungen zu einem SNS Amazon-Thema veröffentlicht werden.
Hochladen einer Archivdatei in den Tresor
Initiieren eines Auftrags zum Abrufen eines Archivs
- Python
-
- SDKfür Python (Boto3)
-
Anmerkung
Es gibt noch mehr dazu. GitHub Sie sehen das vollständige Beispiel und erfahren, wie Sie das AWS -Code-Beispiel-Repository
einrichten und ausführen. Erstellen Sie eine Klasse, die S3-Glacier-Operationen einschließt.
import argparse import logging import os import boto3 from botocore.exceptions import ClientError logger = logging.getLogger(__name__) class GlacierWrapper: """Encapsulates Amazon S3 Glacier API operations.""" def __init__(self, glacier_resource): """ :param glacier_resource: A Boto3 Amazon S3 Glacier resource. """ self.glacier_resource = glacier_resource def create_vault(self, vault_name): """ Creates a vault. :param vault_name: The name to give the vault. :return: The newly created vault. """ try: vault = self.glacier_resource.create_vault(vaultName=vault_name) logger.info("Created vault %s.", vault_name) except ClientError: logger.exception("Couldn't create vault %s.", vault_name) raise else: return vault def list_vaults(self): """ Lists vaults for the current account. """ try: for vault in self.glacier_resource.vaults.all(): logger.info("Got vault %s.", vault.name) except ClientError: logger.exception("Couldn't list vaults.") raise @staticmethod def upload_archive(vault, archive_description, archive_file): """ Uploads an archive to a vault. :param vault: The vault where the archive is put. :param archive_description: A description of the archive. :param archive_file: The archive file to put in the vault. :return: The uploaded archive. """ try: archive = vault.upload_archive( archiveDescription=archive_description, body=archive_file ) logger.info( "Uploaded %s with ID %s to vault %s.", archive_description, archive.id, vault.name, ) except ClientError: logger.exception( "Couldn't upload %s to %s.", archive_description, vault.name ) raise else: return archive @staticmethod def initiate_archive_retrieval(archive): """ Initiates an archive retrieval job. Standard retrievals typically complete within 3—5 hours. When the job completes, you can get the archive contents by calling get_output(). :param archive: The archive to retrieve. :return: The archive retrieval job. """ try: job = archive.initiate_archive_retrieval() logger.info("Started %s job with ID %s.", job.action, job.id) except ClientError: logger.exception("Couldn't start job on archive %s.", archive.id) raise else: return job @staticmethod def list_jobs(vault, job_type): """ Lists jobs by type for the specified vault. :param vault: The vault to query. :param job_type: The type of job to list. :return: The list of jobs of the requested type. """ job_list = [] try: if job_type == "all": jobs = vault.jobs.all() elif job_type == "in_progress": jobs = vault.jobs_in_progress.all() elif job_type == "completed": jobs = vault.completed_jobs.all() elif job_type == "succeeded": jobs = vault.succeeded_jobs.all() elif job_type == "failed": jobs = vault.failed_jobs.all() else: jobs = [] logger.warning("%s isn't a type of job I can get.", job_type) for job in jobs: job_list.append(job) logger.info("Got %s %s job %s.", job_type, job.action, job.id) except ClientError: logger.exception("Couldn't get %s jobs from %s.", job_type, vault.name) raise else: return job_list def set_notifications(self, vault, sns_topic_arn): """ Sets an Amazon Simple Notification Service (Amazon SNS) topic as a target for notifications. Amazon S3 Glacier publishes messages to this topic for the configured list of events. :param vault: The vault to set up to publish notifications. :param sns_topic_arn: The Amazon Resource Name (ARN) of the topic that receives notifications. :return: Data about the new notification configuration. """ try: notification = self.glacier_resource.Notification("-", vault.name) notification.set( vaultNotificationConfig={ "SNSTopic": sns_topic_arn, "Events": [ "ArchiveRetrievalCompleted", "InventoryRetrievalCompleted", ], } ) logger.info( "Notifications will be sent to %s for events %s from %s.", notification.sns_topic, notification.events, notification.vault_name, ) except ClientError: logger.exception( "Couldn't set notifications to %s on %s.", sns_topic_arn, vault.name ) raise else: return notification
Rufen Sie Funktionen in der Wrapper-Klasse auf, um einen Tresor zu erstellen und eine Datei hochzuladen. Konfigurieren Sie dann den Tresor so, dass er Benachrichtigungen veröffentlicht, und initiieren Sie einen Auftrag zum Abrufen des Archivs.
def upload_demo(glacier, vault_name, topic_arn): """ Shows how to: * Create a vault. * Configure the vault to publish notifications to an Amazon SNS topic. * Upload an archive. * Start a job to retrieve the archive. :param glacier: A Boto3 Amazon S3 Glacier resource. :param vault_name: The name of the vault to create. :param topic_arn: The ARN of an Amazon SNS topic that receives notification of Amazon S3 Glacier events. """ print(f"\nCreating vault {vault_name}.") vault = glacier.create_vault(vault_name) print("\nList of vaults in your account:") glacier.list_vaults() print(f"\nUploading glacier_basics.py to {vault.name}.") with open("glacier_basics.py", "rb") as upload_file: archive = glacier.upload_archive(vault, "glacier_basics.py", upload_file) print( "\nStarting an archive retrieval request to get the file back from the " "vault." ) glacier.initiate_archive_retrieval(archive) print("\nListing in progress jobs:") glacier.list_jobs(vault, "in_progress") print( "\nBecause Amazon S3 Glacier is intended for infrequent retrieval, an " "archive request with Standard retrieval typically completes within 3–5 " "hours." ) if topic_arn: notification = glacier.set_notifications(vault, topic_arn) print( f"\nVault {vault.name} is configured to notify the " f"{notification.sns_topic} topic when {notification.events} " f"events occur. You can subscribe to this topic to receive " f"a message when the archive retrieval completes.\n" ) else: print( f"\nVault {vault.name} is not configured to notify an Amazon SNS topic " f"when the archive retrieval completes so wait a few hours." ) print("\nRetrieve your job output by running this script with the --retrieve flag.")
-
APIEinzelheiten finden Sie in der Python-Referenz (Boto3) API in AWS SDK den folgenden Themen.
-