如何使用 Boto3 和 AWS 客戶端的 waiter 功能來檢查 bucket 是否不存在?


問題陳述 − 使用 Python 中的 boto3 庫,使用 waiter 功能檢查 bucket 是否不存在。例如,使用 waiter 檢查 S3 中 Bucket_2 是否不存在。

解決此問題的方法/演算法

步驟 1 − 匯入 boto3 和 botocore 異常以處理異常。

步驟 2 − 使用 bucket_name 作為函式中的引數。

步驟 3 − 使用 boto3 庫建立 AWS 會話。

步驟 4 − 為 S3 建立 AWS 客戶端。

步驟 5 − 現在使用 get_waiter 函式建立 bucket_not_exists 的等待物件。

步驟 6 − 現在,使用 wait 物件驗證 bucket 是否不存在。預設情況下,它每 5 秒檢查一次,直到達到成功狀態。這裡成功狀態意味著 bucket 不應該存在於 S3 中。在 20 次失敗檢查後返回錯誤。但是,使用者可以定義輪詢時間和最大嘗試次數。

步驟 7 − 它返回 None。

步驟 8 − 如果在檢查 bucket 期間出現任何錯誤,請處理通用異常。

示例

使用以下程式碼使用 waiter 檢查 bucket 是否不存在 −

import boto3
from botocore.exceptions import ClientError

def use_waiters_check_bucket_not_exists(bucket_name):
   session = boto3.session.Session()
   s3_client = session.client('s3')
   try:
      waiter = s3_client.get_waiter('bucket_not_exists')
      waiter.wait(Bucket=bucket_name,
                  WaiterConfig={
                     'Delay': 2, 'MaxAttempts': 5})
      print('Bucket not exist: ' + bucket_name)
   except ClientError as e:
      raise Exception( "boto3 client error in use_waiters_check_bucket_not_exists: " + e.__str__())
   except Exception as e:
      raise Exception( "Unexpected error in use_waiters_check_bucket_not_exists: " + e.__str__())

print(use_waiters_check_bucket_not_exists("Bucket_2"))
print(use_waiters_check_bucket_not_exists("Bucket_1"))

輸出

Bucket not exist: Bucket_2
None

botocore.exceptions.WaiterError: Waiter BucketNotExists failed: Max
attempts exceeded
"Unexpected error in use_waiters_check_bucket_not_exists: " +
e.__str__())
Exception: Unexpected error in use_waiters_check_bucket_not_exists:
Waiter BucketNotExists failed: Max attempts exceed

對於 Bucket_2,輸出是 print 語句和 None。由於響應沒有返回任何內容,因此它列印 None。

對於 Bucket_1,輸出是一個異常,因為即使在最大嘗試次數檢查後,此 bucket 仍然存在。

在異常中,可以讀取最大嘗試次數已超過。

更新於: 2021-03-22

923 次檢視

啟動您的 職業生涯

透過完成課程獲得認證

開始學習
廣告