文件 AWS 開發套件範例 GitHub 儲存庫中有更多可用的 [AWS SDK 範例](https://github.com/awsdocs/aws-doc-sdk-examples)。
本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
# 使用適用於 Python 的 SDK (Boto3) 的 Amazon Glacier 範例
下列程式碼範例示範如何使用 適用於 Python (Boto3) 的 AWS SDK 搭配 Amazon Glacier 執行動作和實作常見案例。
*Actions* 是大型程式的程式碼摘錄，必須在內容中執行。雖然動作會告訴您如何呼叫個別服務函數，但您可以在其相關情境中查看內容中的動作。
*案例*是向您展示如何呼叫服務中的多個函數或與其他 AWS 服務組合來完成特定任務的程式碼範例。
每個範例均包含完整原始碼的連結，您可在連結中找到如何設定和執行內容中程式碼的相關指示。
**Topics**
+ [動作](#actions)
+ [案例](#scenarios)
## 動作
### `CreateVault`
以下程式碼範例顯示如何使用 `CreateVault`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/glacier#code-examples)中設定和執行。
```
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
```
+ 如需 API 詳細資訊，請參閱《*適用於 Python (Boto3) 的AWS 開發套件 API 參考*》中的 [CreateVault](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/CreateVault)。
### `DeleteArchive`
以下程式碼範例顯示如何使用 `DeleteArchive`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/glacier#code-examples)中設定和執行。
```
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
@staticmethod
def delete_archive(archive):
"""
Deletes an archive from a vault.
:param archive: The archive to delete.
"""
try:
archive.delete()
logger.info(
"Deleted archive %s from vault %s.", archive.id, archive.vault_name
)
except ClientError:
logger.exception("Couldn't delete archive %s.", archive.id)
raise
```
+ 如需 API 詳細資訊，請參閱《*適用於 Python (Boto3) 的 AWS 開發套件 API 參考*》中的 [DeleteArchive](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/DeleteArchive)。
### `DeleteVault`
以下程式碼範例顯示如何使用 `DeleteVault`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/glacier#code-examples)中設定和執行。
```
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
@staticmethod
def delete_vault(vault):
"""
Deletes a vault.
:param vault: The vault to delete.
"""
try:
vault.delete()
logger.info("Deleted vault %s.", vault.name)
except ClientError:
logger.exception("Couldn't delete vault %s.", vault.name)
raise
```
+ 如需 API 詳細資訊，請參閱《*適用於 Python (Boto3) 的AWS 開發套件 API 參考*》中的 [DeleteVault](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/DeleteVault)。
### `DeleteVaultNotifications`
以下程式碼範例顯示如何使用 `DeleteVaultNotifications`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/glacier#code-examples)中設定和執行。
```
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
@staticmethod
def stop_notifications(notification):
"""
Stops notifications to the configured Amazon SNS topic.
:param notification: The notification configuration to remove.
"""
try:
notification.delete()
logger.info("Notifications stopped.")
except ClientError:
logger.exception("Couldn't stop notifications.")
raise
```
+ 如需 API 詳細資訊，請參閱《*適用於 Python (Boto3) 的 AWS 開發套件 API 參考*》中的 [DeleteVaultNotifications](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/DeleteVaultNotifications)。
### `DescribeJob`
以下程式碼範例顯示如何使用 `DescribeJob`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/glacier#code-examples)中設定和執行。
```
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
@staticmethod
def get_job_status(job):
"""
Gets the status of a job.
:param job: The job to query.
:return: The current status of the job.
"""
try:
job.load()
logger.info(
"Job %s is performing action %s and has status %s.",
job.id,
job.action,
job.status_code,
)
except ClientError:
logger.exception("Couldn't get status for job %s.", job.id)
raise
else:
return job.status_code
```
+ 如需 API 的詳細資訊，請參閱《*適用於 Python (Boto3) 的AWS 開發套件 API 參考*》中的 [DescribeJob](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/DescribeJob)。
### `GetJobOutput`
以下程式碼範例顯示如何使用 `GetJobOutput`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/glacier#code-examples)中設定和執行。
```
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
@staticmethod
def get_job_output(job):
"""
Gets the output of a job, such as a vault inventory or the contents of an
archive.
:param job: The job to get output from.
:return: The job output, in bytes.
"""
try:
response = job.get_output()
out_bytes = response["body"].read()
logger.info("Read %s bytes from job %s.", len(out_bytes), job.id)
if "archiveDescription" in response:
logger.info(
"These bytes are described as '%s'", response["archiveDescription"]
)
except ClientError:
logger.exception("Couldn't get output for job %s.", job.id)
raise
else:
return out_bytes
```
+ 如需 API 詳細資訊，請參閱《*適用於 Python (Boto3) 的AWS 開發套件 API 參考*》中的 [GetJobOutput](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/GetJobOutput)。
### `GetVaultNotifications`
以下程式碼範例顯示如何使用 `GetVaultNotifications`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/glacier#code-examples)中設定和執行。
```
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
@staticmethod
def get_notification(vault):
"""
Gets the currently notification configuration for a vault.
:param vault: The vault to query.
:return: The notification configuration for the specified vault.
"""
try:
notification = vault.Notification()
logger.info(
"Vault %s notifies %s on %s events.",
vault.name,
notification.sns_topic,
notification.events,
)
except ClientError:
logger.exception("Couldn't get notification data for %s.", vault.name)
raise
else:
return notification
```
+ 如需 API 詳細資訊，請參閱《*適用於 Python (Boto3) 的AWS 開發套件 API 參考*》中的 [GetVaultNotifications](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/GetVaultNotifications)。
### `InitiateJob`
以下程式碼範例顯示如何使用 `InitiateJob`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/glacier#code-examples)中設定和執行。
擷取保存庫庫存。
```
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
@staticmethod
def initiate_inventory_retrieval(vault):
"""
Initiates an inventory retrieval job. The inventory describes the contents
of the vault. Standard retrievals typically complete within 3—5 hours.
When the job completes, you can get the inventory by calling get_output().
:param vault: The vault to inventory.
:return: The inventory retrieval job.
"""
try:
job = vault.initiate_inventory_retrieval()
logger.info("Started %s job with ID %s.", job.action, job.id)
except ClientError:
logger.exception("Couldn't start job on vault %s.", vault.name)
raise
else:
return job
```
從保存庫擷取封存。
```
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
@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
```
+ 如需 API 詳細資訊，請參閱《*適用於 Python (Boto3) 的AWS 開發套件 API 參考*》中的 [InitiateJob](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/InitiateJob)。
### `ListJobs`
以下程式碼範例顯示如何使用 `ListJobs`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/glacier#code-examples)中設定和執行。
```
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
@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
```
+ 如需 API 詳細資訊，請參閱《*AWS SDK for Python (Boto3) API 參考*》中的 [ListJobs](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/ListJobs)。
### `ListVaults`
以下程式碼範例顯示如何使用 `ListVaults`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/glacier#code-examples)中設定和執行。
```
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 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
```
+ 如需 API 詳細資訊，請參閱《*適用於 Python (Boto3) 的AWS 開發套件 API 參考*》中的 [ListVaults](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/ListVaults)。
### `SetVaultNotifications`
以下程式碼範例顯示如何使用 `SetVaultNotifications`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/glacier#code-examples)中設定和執行。
```
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 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
```
+ 如需 API 詳細資訊，請參閱《*適用於 Python (Boto3) 的AWS 開發套件 API 參考*》中的 [SetVaultNotifications](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/SetVaultNotifications)。
### `UploadArchive`
以下程式碼範例顯示如何使用 `UploadArchive`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/glacier#code-examples)中設定和執行。
```
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
@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
```
+ 如需 API 的詳細資訊，請參閱《*適用於 Python (Boto3) 的AWS 開發套件 API 參考*》中的 [UploadArchive](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/UploadArchive)。
## 案例
### 封存檔案、取得通知並啟動工作
以下程式碼範例顯示做法：
+ 建立 Amazon Glacier 保存庫。
+ 設定保存庫，以將通知發布至 Amazon SNS 主題。
+ 將封存檔上傳至保存庫。
+ 啟動封存擷取任務。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/glacier#code-examples)中設定和執行。
建立包裝 Amazon Glacier 操作的類別。
```
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
```
呼叫包裝函式類別上的函數，以建立保存庫並上傳檔案，然後將保存庫設定為發布通知與啟動工作以擷取封存。
```
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.")
```
+ 如需 API 詳細資訊，請參閱《適用於 Python (Boto3) 的AWS SDK API 參考》**中的下列主題。
+ [CreateVault](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/CreateVault)
+ [InitiateJob](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/InitiateJob)
+ [ListJobs](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/ListJobs)
+ [ListVaults](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/ListVaults)
+ [SetVaultNotifications](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/SetVaultNotifications)
+ [UploadArchive](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/UploadArchive)
### 取得封存內容並刪除封存
以下程式碼範例顯示做法：
+ 列出 Amazon Glacier 保存庫的任務並取得任務狀態。
+ 取得已完成封存擷取工作的輸出。
+ 刪除封存。
+ 刪除保存庫。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/glacier#code-examples)中設定和執行。
建立包裝 Amazon Glacier 操作的類別。
```
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
@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
@staticmethod
def get_job_output(job):
"""
Gets the output of a job, such as a vault inventory or the contents of an
archive.
:param job: The job to get output from.
:return: The job output, in bytes.
"""
try:
response = job.get_output()
out_bytes = response["body"].read()
logger.info("Read %s bytes from job %s.", len(out_bytes), job.id)
if "archiveDescription" in response:
logger.info(
"These bytes are described as '%s'", response["archiveDescription"]
)
except ClientError:
logger.exception("Couldn't get output for job %s.", job.id)
raise
else:
return out_bytes
@staticmethod
def delete_archive(archive):
"""
Deletes an archive from a vault.
:param archive: The archive to delete.
"""
try:
archive.delete()
logger.info(
"Deleted archive %s from vault %s.", archive.id, archive.vault_name
)
except ClientError:
logger.exception("Couldn't delete archive %s.", archive.id)
raise
@staticmethod
def delete_vault(vault):
"""
Deletes a vault.
:param vault: The vault to delete.
"""
try:
vault.delete()
logger.info("Deleted vault %s.", vault.name)
except ClientError:
logger.exception("Couldn't delete vault %s.", vault.name)
raise
```
呼叫包裝函式類別的函數，從已完成的工作取得封存內容，然後刪除封存。
```
def retrieve_demo(glacier, vault_name):
"""
Shows how to:
* List jobs for a vault and get job status.
* Get the output of a completed archive retrieval job.
* Delete an archive.
* Delete a vault.
:param glacier: A Boto3 Amazon S3 Glacier resource.
:param vault_name: The name of the vault to query for jobs.
"""
vault = glacier.glacier_resource.Vault("-", vault_name)
try:
vault.load()
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceNotFoundException":
print(
f"\nVault {vault_name} doesn't exist. You must first run this script "
f"with the --upload flag to create the vault."
)
return
else:
raise
print(f"\nGetting completed jobs for {vault.name}.")
jobs = glacier.list_jobs(vault, "completed")
if not jobs:
print("\nNo completed jobs found. Give it some time and try again later.")
return
retrieval_job = None
for job in jobs:
if job.action == "ArchiveRetrieval" and job.status_code == "Succeeded":
retrieval_job = job
break
if retrieval_job is None:
print(
"\nNo ArchiveRetrieval jobs found. Give it some time and try again "
"later."
)
return
print(f"\nGetting output from job {retrieval_job.id}.")
archive_bytes = glacier.get_job_output(retrieval_job)
archive_str = archive_bytes.decode("utf-8")
print("\nGot archive data. Printing the first 10 lines.")
print(os.linesep.join(archive_str.split(os.linesep)[:10]))
print(f"\nDeleting the archive from {vault.name}.")
archive = glacier.glacier_resource.Archive(
"-", vault.name, retrieval_job.archive_id
)
glacier.delete_archive(archive)
print(f"\nDeleting {vault.name}.")
glacier.delete_vault(vault)
```
+ 如需 API 詳細資訊，請參閱《適用於 Python (Boto3) 的AWS SDK API 參考》**中的下列主題。
+ [DeleteArchive](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/DeleteArchive)
+ [DeleteVault](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/DeleteVault)
+ [GetJobOutput](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/GetJobOutput)
+ [ListJobs](https://docs.aws.amazon.com/goto/boto3/glacier-2012-06-01/ListJobs)