How to invoke retry mechanism of botocore? #3283
Unanswered
budakoti-mohit
asked this question in
Q&A
Replies: 0 comments
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
-
I am using following code:
import boto3
import botocore
from botocore.config import Config
import logging
Enable logging
logging.basicConfig(level=logging.DEBUG)
config = Config(
retries={'max_attempts': 5}
)
s3 = boto3.client('s3', config=config, endpoint_url='http://localhost:4566')
Intentionally return transient errors
def list_buckets_with_transient_errors():
raise botocore.exceptions.ClientError(
error_response={
'Error': {
'Code': 'Throttling',
'Message': 'Rate exceeded'
},
'ResponseMetadata': {
'HTTPStatusCode': 429
}
},
operation_name='ListBuckets'
)
Patch the list_buckets method to return transient errors
original_list_buckets = s3.list_buckets
s3.list_buckets = list_buckets_with_transient_errors
try:
s3.list_buckets()
except botocore.exceptions.ClientError as e:
logging.info(f"Final error after retries: {e}")
Restore the original list_buckets method
s3.list_buckets = original_list_buckets
I am running localstack docker onto my local with the endpoint_url. Wanted to see the logs with delay and max_attempts should be consumed upto the total limit passed & print Final error after retries
Beta Was this translation helpful? Give feedback.
All reactions