如何在 Boto3 中使用 Wait 功能驗證 S3 桶中是否存在某個鍵?
問題陳述 − 使用 Python 中的 boto3 庫,利用等待器功能檢查桶中是否存在某個鍵。例如,使用等待器檢查 Bucket_1 中是否存在名為 test1.zip 的鍵。
解決此問題的方法/演算法
步驟 1 − 匯入 boto3 和 botocore 異常以處理異常。
步驟 2 − bucket_name 和 key 是函式中的兩個引數。
步驟 3 − 使用 boto3 庫建立一個 AWS 會話。
步驟 4 − 為 S3 建立一個 AWS 客戶端。
步驟 5 − 現在使用 get_waiter 函式為 object_not_exists 建立等待物件。
步驟 6 − 現在,使用等待物件驗證給定桶中是否存在鍵。預設情況下,它每 5 秒檢查一次,直到達到成功狀態。在 20 次失敗檢查後返回錯誤。但是,使用者可以定義輪詢時間和最大嘗試次數。
步驟 7 − 它返回 None。
步驟 8 − 如果在檢查桶時出現錯誤,則處理通用異常。
示例
使用以下程式碼使用等待器檢查桶中是否存在鍵:
import boto3
from botocore.exceptions import ClientError
def use_waiters_check_object_not_exists(bucket_name, key_name):
session = boto3.session.Session()
s3_client = session.client('s3')
try:
waiter = s3_client.get_waiter('object_not_exists')
waiter.wait(Bucket=bucket_name, Key = key_name,
WaiterConfig={
'Delay': 2, 'MaxAttempts': 5})
print('Object does not exist: ' + bucket_name +'/'+key_name)
except ClientError as e:
raise Exception( "boto3 client error in use_waiters_check_object_not_exists: " + e.__str__())
except Exception as e:
raise Exception( "Unexpected error in use_waiters_check_object_not_exists: " + e.__str__())
print(use_waiters_check_object_exists("Bucket_1","testfolder/test1.zip"))
print(use_waiters_check_object_exists("Bucket_1","testfolder/test.zip"))輸出
Object does not exist: Bucket_1/testfolder/test1.zip None botocore.exceptions.WaiterError: Waiter ObjectNotExists failed: Max attempts exceeded "Unexpected error in use_waiters_check_object_not_exists: " + e.__str__()) Exception: Unexpected error in use_waiters_check_object_not_exists: Waiter ObjectNotExists failed: Max attempts exceed
對於 Bucket_1/testfolder/test1.zip,輸出是列印語句和 None。由於響應沒有返回任何內容,因此它列印 None。
對於 Bucket_1/testfolder/test.zip,由於此物件存在,因此輸出是一個異常。
在異常中,可以讀取“最大嘗試次數已超過”。
廣告
資料結構
網路
關係資料庫管理系統 (RDBMS)
作業系統
Java
iOS
HTML
CSS
Android
Python
C語言程式設計
C++
C#
MongoDB
MySQL
Javascript
PHP