Correctly managing postgresql connections in celery task for Flask-SQLAlchemy and Celery

Question

I'm using Flask-SQLAlchemy, Celery and uWSGI.

I know that Flask-SQLAlchemy automatically manages the session for you. I'm not sure how this works with Celery workers, but it seems that when I run a task a second time, I get the following error: DatabaseError: (psycopg2.DatabaseError) server closed the connection unexpectedly.

Here's how I create the app context and celery tasks:

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

It seems that maybe the workers are using the same database connection and after a task completes that connection is not replenished?

It may be related to the following question?

I'm not sure how to correctly setup the workers or celery so that they're using new connections to the database..

claudio claudio · Accepted Answer · 2016-06-25T00:24:32

Okay. I figured it out, for every process that's using an application context, you must use a new application context. Before, in my app/__init__.py I was simply creating the application globally like so:

from flask import Flask
app = Flask(__name__)

I then changed my app to use create_app like in this pattern

Now, my tasks.py looks like this:

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

Make sure in your create_app you are calling db.init_app(app).

Correctly managing postgresql connections in celery task for Flask-SQLAlchemy and Celery

1 Answers