使用 Flask 和 SQLAlchemy 在 Celery 任务中未更新数据库

Question

我正在使用 Flask 和 SQLAlchemy 编写 Web 应用程序。我的程序需要在后台处理一些内容，然后将这些内容标记为在数据库中已处理。使用标准 Flask/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):
    # process stuff here

    stuff.processed = True
    db.session.commit()

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

我可以从

process_stuff

访问我的数据库（例如提交类似

Stuff.query.get(some_id)

工作的查询），但

db.session.commit()

什么也不做：我的

stuff

记录未更新。根据 Celery 工作日志，发生了提交，但数据库中没有任何变化。我的

db.session.commit()

有什么问题吗？是否有可能以某种方式进行这样的提交？

Answer 1

好的，我明白了。传递给

stuff

的

process_stuff()

未附加到

db.session

。我必须在

process_stuff()

中提出明确的请求才能获得正确的

stuff

对象，如下所示：

@celery.task()
def process_stuff(stuff):
    # process stuff here

    my_stuff = Stuff.query.get(stuff.id)

    my_stuff.processed = True
    db.session.commit()

现在可以了。

Answer 2

我有一个非常相似的问题。我以为我已经正确地将传入的对象绑定到 db.session，但事实证明我需要在将其传入之前将其与上一个会话解除绑定。因此，在将其传入之前，我必须将行 db.session 放入其中。 session.expunge（东西）。