Flask, progetti utilizza compito sedano e ottenuto ciclo di importazione

Question

Ho un'applicazione con Blueprints e sedano il codice è qui:

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: os importazione

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: da App importazione make_celery, ap p_conf

cel = make_celery(app_conf.app) 

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

e qui è il problema: The Blueprint "messa in scena" usa compito send_realm_to_fabricdb, in modo che rende: from tasks import send_realm_to_fabricdb che, quando ho appena eseguire l'applicazione, tutto va bene Ma, quando sto cercando di eseguire sedano: celery -A app.tasks worker -l info --beat, va a cel = make_celery(app_conf.app) in tasks.py, ottenuto app = Nessuno e cercando di creare nuovamente l'applicazione: registrazione di un progetto ... quindi ho importato il ciclo qui. Potresti dirmi come interrompere questo ciclo? Grazie in anticipo.

Answer 1

Non ho il codice per provare questo, ma penso che le cose funzionerebbero meglio se spostate la creazione dell'istanza di Celery da tasks.py e nella funzione create_app, in modo che avvenga nello stesso tempo L'istanza app viene creata.

L'argomento si dà al lavoratore Sedano nell'opzione -A non ha bisogno di avere i compiti, sedano bisogno solo l'oggetto sedano, così per esempio, è possibile creare uno script di avviamento a parte, dire celery_worker.py che chiama create_app per creare app e cel e quindi consegnarlo al lavoratore come -A celery_worker.cel, senza coinvolgere affatto il progetto.

Spero che questo aiuti.