Very often, 3rd party API clients are designed to return a generic "API Exception", with details contained inside some property. One notorious example of this is boto3: most exceptions come as a ClientError with a response.
Usually this is fine, but occasionally I need finer-grained exceptions than what the client is giving me.
One common scenario is configuring a celery task to retry a while after the rate limiting: I want to get errors for all exceptions, but for rate-limiting, I just want to retry the task later. Celery offers a convenient autoretry_for argument for that, but we can't use it just for throttling because boto3 does not return an exception specific enough.
I could wrap the login in a try ... except clause and inspect the exception, but that get repetitive pretty quickly, especially as I add more and more tasks.
For these kind of situations, I create a context decorator. The decorator inspects the exception for me and, if it's the one I'm looking, raises a custom exception that then I can catch:
# myproject/exceptions.py import contexlib from botocore.exceptions import ClientError class ThrottleException(ClientError): pass class raise_throttle(contextlib.ContextDecorator): def __enter__(self): return self def __exit__(self, exc_type, exc_value, tb): if exc_type == ClientError: if exc_value.response.get("Error", {}).get("Code") == "Throttling": raise ThrottleException( exc_value.response, exc_value.operation_name ) from exc_value # returning `False` makes the decorator # raise the original exception, if any. return False This decorator can be used on any function or method that calls the API:
# myproject/myapp/aws.py import boto3 @raise_throttle() def upload(content): s3 = boto3.resource("s3") bucket = s3.Bucket("my-bucket") bucket.put_object( Body=b"lorem ipsum", Key="Hamlet.txt", ) And the exception can be caught by celery's autoretry_for:
# myproject/myapp/tasks.py from myproject.celery import app from myproject.exceptions import ThrottleException from myproject.myapp.aws import upload @app.task(autoretry_for=(ThrottleException,)) def my_task(content): upload(content) But I don't want to retry the task immediately, since the throttling takes a while to be lifted. Therefore I create a Task base class that retries with exponential backoff:
# myproject/tasks.py import random from celery import Task def jitter(jitter_max=1.4): return random.uniform(1, jitter_max) def exponential(retries, factor=3): return (factor ** (retries + 1)) def exp_jitter(retries, exp_factor=3, jitter_max=1.4): return exponential(retries, exp_factor) * jitter(jitter_max) class ExpBackoffTask(Task): abstract = True max_retries = 10 def retry(self, *args, **kwargs): countdown = kwargs.get('countdown', None) if countdown is None: # if no explicit countdown is given, use an exponential backoff # with some random jitter, giving us a top-end under 24 hours at # which the last retry will be attempted. kwargs['countdown'] = int( exp_jitter(self.request.retries) ) return super().retry(*args, **kwargs) I can then use the custom class as base:
# myproject/myapp/tasks.py from myproject.celery import app from myproject.exceptions import ThrottleException from myproject.tasks import ExpBackoffTask from myproject.myapp.aws import upload @app.task(base=ExpBackoffTask, autoretry_for=(ThrottleException,)) def my_task(content): upload(content) 
Top comments (0)