Flask, verwendet Baupläne Sellerie Aufgabe und bekam Zyklus Import

Question

Ich habe eine Anwendung mit Riss und Sellerie der Code ist hier:

config.py

import os 
from celery.schedules import crontab 
basedir = os.path.abspath(os.path.dirname(__file__)) 

class Config: 
    SECRET_KEY = os.environ.get('SECRET_KEY') or '' 
    SQLALCHEMY_COMMIT_ON_TEARDOWN = True 
    RECORDS_PER_PAGE = 40 
    SQLALCHEMY_DATABASE_URI = '' 
    CELERY_BROKER_URL = '' 
    CELERY_RESULT_BACKEND = '' 
    CELERY_RESULT_DBURI = '' 
    CELERY_TIMEZONE = 'Europe/Kiev' 
    CELERY_ENABLE_UTC = False 
    CELERYBEAT_SCHEDULE = {} 

    @staticmethod 
    def init_app(app): 
     pass 


class DevelopmentConfig(Config): 
    DEBUG = True 
    WTF_CSRF_ENABLED = True 
    APP_HOME = '' 
    SQLALCHEMY_DATABASE_URI = 'mysql+mysqldb://...' 
    CELERY_BROKER_URL = 'sqla+mysql://...' 
    CELERY_RESULT_BACKEND = "database" 
    CELERY_RESULT_DBURI = 'mysql://...' 
    CELERY_TIMEZONE = 'Europe/Kiev' 
    CELERY_ENABLE_UTC = False 
    CELERYBEAT_SCHEDULE = { 
     'send-email-every-morning': { 
      'task': 'app.workers.tasks.send_email_task', 
      'schedule': crontab(hour=6, minute=15), 
     }, 
    } 


class TestConfig(Config): 
    DEBUG = True 
    WTF_CSRF_ENABLED = False 
    TESTING = True 
    SQLALCHEMY_DATABASE_URI = 'mysql+mysqldb://...' 


class ProdConfig(Config): 
    DEBUG = False 
    WTF_CSRF_ENABLED = True 
    SQLALCHEMY_DATABASE_URI = 'mysql+mysqldb://...' 
    CELERY_BROKER_URL = 'sqla+mysql://...celery' 
    CELERY_RESULT_BACKEND = "database" 
    CELERY_RESULT_DBURI = 'mysql://.../celery' 
    CELERY_TIMEZONE = 'Europe/Kiev' 
    CELERY_ENABLE_UTC = False 
    CELERYBEAT_SCHEDULE = { 
     'send-email-every-morning': { 
      'task': 'app.workers.tasks.send_email_task', 
      'schedule': crontab(hour=6, minute=15), 
     }, 
    } 

config = { 
    'development': DevelopmentConfig, 
    'default': ProdConfig, 
    'production': ProdConfig, 
    'testing': TestConfig, 
} 


class AppConf: 
    """ 
    Class to store current config even out of context 
    """ 
    def __init__(self): 
     self.app = None 
     self.config = {} 

    def init_app(self, app): 
     if hasattr(app, 'config'): 
      self.app = app 
      self.config = app.config.copy() 
     else: 
      raise TypeError 

init Py: import os

from flask import Flask 
from celery import Celery 
from config import config, AppConf 

def create_app(config_name): 
    app = Flask(__name__) 
    app.config.from_object(config[config_name]) 
    config[config_name].init_app(app) 
    app_conf.init_app(app) 

    # Connect to Staging view 
    from staging.views import staging as staging_blueprint 
    app.register_blueprint(staging_blueprint) 

    return app 


def make_celery(app=None): 
    app = app or create_app(os.getenv('FLASK_CONFIG') or 'default') 
    celery = Celery(__name__, broker=app.config.CELERY_BROKER_URL) 
    celery.conf.update(app.conf) 
    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 

tasks.py: von app import make_celery, ap p_conf

cel = make_celery(app_conf.app) 

@cel.task 
def send_realm_to_fabricdb(realm, form): 
    some actions... 

und hier ist das Problem: The Blueprint "staging" verwendet Aufgabe send_realm_to_fabricdb, so macht es: from tasks import send_realm_to_fabricdb als, wenn ich nur Anwendung ausführen, geht alles in Ordnung ABER, wenn ich versuche, Run Sellerie: celery -A app.tasks worker -l info --beat, geht es zu cel = make_celery(app_conf.app) in tasks.py, habe app = None und versuche, Anwendung neu zu erstellen: Registrierung eines Blueprint ... so habe ich Zyklus importieren hier. Können Sie mir sagen, wie man diesen Zyklus bricht? Vielen Dank im Voraus.

Answer 1

Ich habe nicht den Code, um dies auszuprobieren, aber ich denke, Dinge würden besser funktionieren, wenn Sie die Erstellung der Sellerie-Instanz aus tasks.py und in die create_app Funktion verschieben, so dass es zur gleichen Zeit passiert app Instanz wird erstellt.

Das Argument Sie in der -A Option zum Sellerie Arbeiter geben muss nicht die Aufgaben haben, Sellerie Sellerie Objekt gerade braucht, so zum Beispiel ein separates Startskript sagen celery_worker.py schaffen könnte, die create_app ruft zu erstellen app und cel und geben Sie es dann dem Arbeiter als -A celery_worker.cel, ohne die Blaupause überhaupt zu beteiligen.

Hoffe, das hilft.