带有 Flask 和 Flask-SQLAlchemy 的 Celery 无法更新数据库记录答案

【问题标题】：Celery with Flask and Flask-SQLAlchemy cannot update a database record带有 Flask 和 Flask-SQLAlchemy 的 Celery 无法更新数据库记录
【发布时间】：2017-12-15 13:37:25
【问题描述】：

我正在使用 Flask 和 Flask-SQLAlchemy 创建一个后台 celery 任务来更新数据库记录的属性。我使用 recommended documentation 进行 celery 配置，这是我的代码的简化版本：

from flask import Flask
from celery import Celery

def make_celery(app):
    celery = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)
    TaskBase = celery.Task
    class ContextTask(TaskBase):
        abstract = True
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
    celery.Task = ContextTask
    return celery


app = Flask(__name__)

celery = make_celery(app)

class Stuff(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    processed = db.Column(db.Boolean)


@celery.task()
def process_stuff(stuff_id):
    stuff = Stuff.query.get(stuff_id)
    print("stuff.processed 1: {}".format(stuff.processed))
    stuff.processed = True
    print("stuff.processed 2: {}".format(stuff.processed))
    db.session.add(stuff)
    db.session.commit()
    print("stuff.processed 3: {}".format(stuff.processed))

@app.route("/process_stuff/<id>")
def do_process_stuff(id):
    stuff = Stuff.query.get_or_404(id)
    process_stuff.delay(stuff.id)
    return redirect(url_for("now_wait"))

这是打印语句的输出：

[2017-07-11 07:32:01,281: WARNING/PoolWorker-4] stuff.processed 1: False
[2017-07-11 07:32:01,282: WARNING/PoolWorker-4] stuff.processed 2: False
[2017-07-11 07:32:01,285: WARNING/PoolWorker-4] stuff.processed 3: False

我可以在我的 celery worker 日志中看到任务正在接受并完成；但是，打印语句显示 stuff.processed 属性始终为 False - 即使在我手动更新它之后，打印语句也永远不会显示 True（我已经在 celery 之外对此进行了测试，我可以很好地更新属性）。

有一个非常相似的问题here，但该解决方案不适用于我的情况

库版本

烧瓶 0.12.2
SQLAlchemy 1.1.11
芹菜 4.0.2
Flask-SQLAlchemy 2.2

更新

其他测试表明我可以创建新创建的对象并将其保存到数据库中 - 更新预先存在的对象仍然失败。

【问题讨论】：

尝试将方法def on_success(self, retval, task_id, args, kwargs): db.session.commit()添加到ContextTask
@DanilaGanchar 不幸的是没有运气
你能在问题中添加烧瓶、芹菜、sqlalchemy 版本吗？顺便说一句，我不能重复你的行为。
@DanilaGanchar 没问题，我已经包含了这些库。我也在使用 postgresql 作为数据库
如果你使用 Flask-SQLAlchemy，你需要为你的 celery 任务中的数据库调用创建一个请求上下文。虽然我不确定为什么你没有收到一个错误说明（除非有其他东西隐藏了异常）

标签： python flask sqlalchemy celery flask-sqlalchemy

【解决方案1】：

在我的问题中，我大大简化了代码，并在这样做时将reserved 属性更改为processed。我的实际模型看起来更像这样：

class Stuff(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    reserved = db.Column(db.Boolean)

原来reserved 是 celery 中的一个函数，这就是为什么无法从我的 celery 任务中更新该属性的原因。解决方案是重命名模型上的reserved 属性。

【讨论】：