Kolba, plany korzysta z selera i dostał zadanie cyklu import

Question

Mam aplikacji z odbitek i selera kod jest tutaj:

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 

startowych .py: os import

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: z aPP importu make_celery AP p_conf

cel = make_celery(app_conf.app) 

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

i tutaj jest problem: Blueprint "staging" używa send_realm_to_fabricdb zadania, co oznacza, że: from tasks import send_realm_to_fabricdb niż, kiedy wystarczy uruchomić aplikację, wszystko idzie ok ale gdy próbuję uruchom selera: celery -A app.tasks worker -l info --beat, przechodzi do cel = make_celery(app_conf.app) w tasks.py, dostał app = None i próbuje ponownie utworzyć aplikację: rejestracja projektu ... więc mam tutaj import cyklu. Czy możesz mi powiedzieć, jak przerwać ten cykl? Z góry dzięki.

Answer 1

nie mam kodu na to wypróbować, ale myślę, że wszystko będzie działać lepiej, jeśli przesunąć utworzenie instancji naciowego z tasks.py i do funkcji create_app, tak, że dzieje się w tym samym czasie Utworzono instancję app.

Argument dać pracownikowi Seler w opcji -A nie trzeba mieć zadania, Seler prostu potrzebuje obiektu seler, tak na przykład, można utworzyć osobny skrypt startowy, powiedzmy celery_worker.py który wywołuje create_app stworzyć app i cel, a następnie przekazać go pracownikowi jako -A celery_worker.cel, bez angażowania całego projektu.

Mam nadzieję, że to pomoże.