python - Celery 中的异常处理？

Question

我有一组任务，每个任务都向标准 oauth API 端点发出请求，并且依赖于 Bearer_token .如果 ，任务将在响应处理期间引发异常。 Bearer_token 已过期。还有一个refresh_bearer_token任务，它在 token 过期后处理更新 token 。

这是一个伪代码:

from proj.celery import app

bearer_token = '1234'

class OauthError(Exception):
    pass

@app.task
def api_request():
    response = request(bearer_token, ...)
    if response.bearer_token_expired:
        raise OauthError('oauth')

@app.task
def refresh_bearer_token():
    ...

我如何安排 refresh_bearer_token每当 OauthError 时执行的任务被提出？

我能找到的唯一解决方案是使用 link_error像这样的夸格:

@app.task
def error_callback(uuid):
    exception_msg = AsyncResult(uuid).get(propagate=False, disable_sync_subtasks=False)
    if exception_msg = 'oauth':
        refresh_bearer_token.delay()
    else:
        raise 

api_request.apply_async(link_error=error_callback.s())

但这似乎不是最优的，原因有几个，最显着的是它在另一个同步子任务中生成一个同步子任务， strongly discourged在文档中。

Is there a more pythonic way of exception catching in celery?

例如:

def catch(func_that_requires_oauth):
    try:
        func_that_requires_oauth.delay()
    except OauthError:
        refresh_bearer_token.delay() | func_that_requires_oauth.delay()

Answer 1

只是在那里抛出一些想法。您可以创建一个基本任务，在调用 refresh_bearer_token 任务时等待或重试锁定。当它失败时，它会启动 refresh_bearer_token 任务并自行退休。

重试将正在运行的任务的副本放到队列的后面

现在你必须实现某种锁定，如果已经获得了锁定，那么 refresh_bearer_token 什么也不做，因为另一个任务应该更新它。您还需要为此“锁定”添加 TTL 以防止出现 refresh_bearer_token 时出现某些情况。任务失败

@app.task
def refresh_bearer_token():
    try:
        with aquire_lock(timeout=0):
            refresh_token()
    except TimeoutError:
        pass


class RequiresOauthTask(app.Task):
    abstract = True

    def __call__(self, *args, **kwargs):
        if lock_is_present():
            self.retry()  # or wait?
        return super().__call__(*args, **kwargs)

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        if isinstance(exc, OauthError):
            refresh_bearer_token.delay()
            self.retry()
        super().on_failure(exc, task_id, args, kwargs, einfo)


@app.task(base=RequiresOauthTask)
def my_task():
    pass