如何使用 Wait 功能來檢查 S3 儲存桶中是否存在某個鍵,使用 Boto3 和 AWS 客戶端?


當用戶希望使用 wait 功能來驗證程式程式碼中儲存桶中是否存在某個鍵時。

問題陳述 - 使用 Python 中的 boto3 庫來檢查儲存桶中是否存在某個鍵,使用 waiters 功能。例如,使用 waiters 檢查 Bucket_1 中是否存在鍵 test.zip。

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

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

步驟 2 - bucket_name 和 key 是函式中的兩個引數。

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

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

步驟 5 - 現在使用 get_waiter 函式為 object_exists 建立 wait 物件。

步驟 6 - 現在,使用 wait 物件來驗證給定儲存桶中是否存在鍵。預設情況下,它每 5 秒檢查一次,直到達到成功狀態。在 20 次失敗檢查後返回錯誤。但是,使用者可以定義輪詢時間和最大嘗試次數。

步驟 7 - 它返回 None。

步驟 8 - 如果在檢查儲存桶時出現問題,則處理通用異常。

示例

使用以下程式碼使用 waiter 來檢查儲存桶中是否存在某個鍵 -

import boto3
from botocore.exceptions import ClientError

def use_waiters_check_object_exists(bucket_name, key_name):
   session = boto3.session.Session()
   s3_client = session.client('s3')
   try:
      waiter = s3_client.get_waiter('object_exists')
      waiter.wait(Bucket=bucket_name, Key = key_name,
                  WaiterConfig={
                     'Delay': 2, 'MaxAttempts': 5})
      print('Object exists: ' + bucket_name +'/'+key_name)
   except ClientError as e:
      raise Exception( "boto3 client error in use_waiters_check_object_exists: " + e.__str__())
   except Exception as e:
      raise Exception( "Unexpected error in use_waiters_check_object_exists: " + e.__str__())

print(use_waiters_check_object_exists("Bucket_1","testfolder/test.zip"))
print(use_waiters_check_object_exists("Bucket_1","testfolder/test1.zip")
)

輸出

Object exists: Bucket_1/testfolder/test.zip
None

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

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

對於 Bucket_1/testfolder/test1.zip,輸出是異常,因為此物件不存在。

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

更新於: 2021年3月22日

2K+ 次檢視

開啟你的 職業生涯

透過完成課程獲得認證

開始學習
廣告

© . All rights reserved.