Очень часто 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”