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
Uh oh!
There was an error while loading. Please reload this page.
-
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