Очень часто 3-я сторонние API-клиенты предназначены для возврата общего «исключения API», с деталями, содержащимися внутри некоторой собственности. Один пресловутый пример этого – BOTO3.
: большинство исключений приходят как ClienterRor
с ответ
Отказ
Обычно это нормально, но иногда мне нужно тонкие исключения, чем то, что клиент дает мне.
Один общий сценарий настраивает задачу сельдерея, чтобы повторить попытку после ограничения скорости: я хочу получить ошибки для всех исключений, но для ограничения курсов я просто хочу повторить задачу позже. Сельрере предлагает удобный Autoretry_for
аргумент для этого, но мы не можем использовать его просто для дросселирования, потому что BOTO3
не возвращает исключение достаточно конкретно.
Я мог бы обернуть логин в попробуй ... кроме
оговорка и проверьте исключение, но это довольно быстро получают повторение, особенно как я добавляю все больше и больше задач.
Для таких ситуаций я создаю контекстный декоратор . Декоратор проверяет исключение для меня и, если это тот, который я смотрю, повышает пользовательское исключение, что тогда я могу поймать:
# 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
Этот декоратор можно использовать на любой функции или методе, который вызывает 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", )
И исключение может быть поймано сельдереем 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)
Но я не хочу немедленно повторить задачу, так как дроссель занимает некоторое время, чтобы быть поднятым. Поэтому я создаю Базовый класс задач Это повторяется с экспоненциальным откровением:
# 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)
Затем я могу использовать пользовательский класс в качестве основания:
# 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)
Оригинал: “https://dev.to/fcurella/refining-exceptions-with-context-decorators-31i5”