正确管理 Flask-SQLAlchemy 和 Celery 的 celery 任务中的 postgresql 连接

Question

我正在使用 Flask-SQLAlchemy、Celery 和 uWSGI。

我知道 Flask-SQLAlchemy 会自动为您管理会话。我不确定这如何与 Celery 工作人员一起使用，但似乎当我第二次运行任务时，出现以下错误：DatabaseError: (psycopg2.DatabaseError) 服务器意外关闭了连接。

以下是我创建应用上下文和 celery 任务的方法：

def make_celery(app):
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_BACKEND'],
        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

似乎工作人员正在使用相同的数据库连接，并且在任务完成后该连接没有得到补充？

可能和following question有关？

我不确定如何正确设置工作人员或 celery，以便他们使用与数据库的新连接..

Answer 1

好的。我想通了，对于每个使用应用程序上下文的进程，您必须使用新的应用程序上下文。之前，在我的 app/__init__.py 中，我只是像这样在全局范围内创建应用程序：

from flask import Flask
app = Flask(__name__)

然后我将我的应用程序更改为使用 create_app 就像在 this pattern 中一样

现在，我的 tasks.py 看起来像这样：

from myapp import create_app
from celery import Celery
def make_celery(app=None):
    app = app or create_app()
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_BACKEND'],
        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

celery = make_celery()

确保在您的 create_app 中您正在调用 db.init_app(app)。