使用Flask和SQLAlchemy的Celery任务中不更新数据库 [英] Database is not updated in Celery task with Flask and SQLAlchemy

问题描述

我正在用Flask和SQLAlchemy编写Web应用程序。我的程序需要在后台处理一些东西，然后将这些东西标记为在数据库中处理。使用标准Flask / Celery示例，我有这样的事情：

from flask import Flags
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 .__调用__（self，* args，** kwargs）
celery.Task = ContextTask
返回芹菜


app = Flask（__ name__）

celery = make_celery（app）

class dbff（db.Model）：
id = db.Column（db.Integer，primary_key = True）
processed = db.Column（db.Boolean）


@ celery.task（）
def process_stuff（ ）
＃处理这里的东西

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） work），但是 db.session.commit（） ：我的东西记录没有更新。根据芹菜工人日志，提交发生，但在数据库中没有任何变化。我的 db.session.commit（）有什么问题吗？是否有可能做出这样的提交？

解决方案

好的，我明白了。 东西传递给 process_stuff（）不附加到 db.session 。我必须在 process_stuff（）中作出明确的请求来得到正确的 stuff 这样的对象：

@ celery.task（）
def process_stuff（stuff）：
＃

my_stuff = Stuff.query.get（stuff.id）

my_stuff.processed = True
db.session.commit（）
code>

现在可以使用了。

I'm writing web application with Flask and SQLAlchemy. My program needs to process some stuff in the background and then mark this stuff as processed in the database. Using standard Flask/Celery example, I have something like this:

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"))

I can access my database from process_stuff (e.g. submit queries like Stuff.query.get(some_id) work), but db.session.commit() do nothing: my stuff record is not updated. According to Celery worker log, commit occures but nothing changes in the database. Is there something wrong with my db.session.commit()? Is it possible to make such commit somehow?

解决方案

Okay, I got it. stuff passed to process_stuff() is not attached to db.session. I have to make explicit request in process_stuff() to get the right stuff object like this:

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

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

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

Now it works.

这篇关于使用Flask和SQLAlchemy的Celery任务中不更新数据库的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！