PostgreSQL-Verbindungen in Sellerie-Task für Flask-SQLAlchemy und Sellerie richtig verwalten

Question

Ich benutze Flask-SQLAlchemy, Sellerie und uWSGI.

Ich weiß, dass Flask-SQLAlchemy automatisch die Sitzung für Sie verwaltet. Ich bin nicht sicher, wie das mit Sellery-Arbeitern funktioniert, aber es scheint, dass wenn ich eine Aufgabe ein zweites Mal ausführe, bekomme ich den folgenden Fehler: DatabaseError: (psycopg2.DatabaseError) Server schloss die Verbindung unerwartet.

Hier ist, wie ich die App Kontext und Sellerie Aufgaben erstellen:

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 

Es scheint, dass vielleicht die Arbeiter die gleiche Datenbank-Verbindung verwenden und nach einer Aufgabe abgeschlossen ist, dass die Verbindung nicht wieder aufgefüllt wird?

Es kann sich auf die following question beziehen?

Ich bin nicht sicher, wie Setup richtig, die Arbeiter oder Sellerie, so dass sie neue Verbindungen zur Datenbank verwenden sind ..

Answer 1

Okay. Ich habe es herausgefunden. Für jeden Prozess, der einen Anwendungskontext verwendet, müssen Sie einen neuen Anwendungskontext verwenden. Früher in meiner app/__init__.py war die Schaffung ich einfach die Anwendung global wie folgt:

from flask import Flask 
app = Flask(__name__) 

änderte ich dann meine app wie zu verwenden create_app in this pattern

Nun, mein tasks.py wie folgt aussieht:

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

Stellen Sie sicher, dass Sie in Ihrer create_app db.init_app (app) aufrufen.