在 Flask 中成功提交后执行 celery 任务？

Question

运行时间相对较长的任务被委派给在另一台服务器上单独运行的 celery worker。

但是，结果被添加回关系数据库（根据task_descr.id 作为键更新表，见下文），worker 使用ignore_result。

从 Flask 应用程序请求的任务：

task = app.celery.send_task('tasks.mytask', [task_descr.id, attachments])

问题是在 Flask 端的事务尚未关闭时请求任务。这会导致竞争条件，因为有时 celery worker 在 Flask 应用程序中的事务结束之前完成任务。

仅在成功交易后发送任务的正确方法是什么？

或者工人应该在尝试有条件的UPDATE之前检查task_descr.id的可用性并重试任务（这感觉太复杂的安排）？

Run function after a certain type of model is committed 的回答讨论了类似的情况，但这里任务发送是明确的，因此不需要在某些模型中监听更新/插入。

Answer 1

其中一种方法是Per-Request After-Request Callbacks，感谢 Armin Ronacher：

from flask import g

def after_this_request(func):
    if not hasattr(g, 'call_after_request'):
        g.call_after_request = []
    g.call_after_request.append(func)
    return func


@app.after_request
def per_request_callbacks(response):
    for func in getattr(g, 'call_after_request', ()):
        response = func(response)
    return response

在我的例子中，用法是嵌套函数的形式：

    task_desc = ...
    attachments = ...
    #...
    @after_this_request
    def send_mytask(response):
        if response.status_code in {200, 302}:
            task = app.celery.send_task('tasks.mytask', [task_descr.id, attachments])
        return response

不理想，但有效。我的任务只针对成功服务的请求，所以我不关心 500s 或其他错误情况。