Flask, blueprints 使用 celery 任务并获得循环导入答案

【问题标题】：Flask, blueprints uses celery task and got cycle importFlask, blueprints 使用 celery 任务并获得循环导入
【发布时间】：2015-06-22 15:18:29
【问题描述】：

我有一个带有蓝图和 Celery 的应用程序代码在这里：

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

初始化.py：导入操作系统

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：从应用导入 make_celery, app_conf

cel = make_celery(app_conf.app)

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

这就是问题所在：蓝图“登台”使用任务 send_realm_to_fabricdb，因此它使：from tasks import send_realm_to_fabricdb 比，当我运行应用程序时，一切正常但是，当我尝试运行 celery 时：celery -A app.tasks worker -l info --beat，它转到 tasks.py 中的cel = make_celery(app_conf.app)，得到 app=None 并再次尝试创建应用程序：注册蓝图...所以我有循环导入这里。你能告诉我如何打破这个循环吗？提前致谢。

【问题讨论】：

Howdie，我目前正在处理同样的问题。你能得到这个工作吗？

标签： import flask celery

【解决方案1】：

我没有代码来尝试这个，但我认为如果你将 Celery 实例的创建从 tasks.py 移到 create_app 函数中，事情会更好，这样它就会发生在同时创建app 实例。

您在-A 选项中提供给 Celery worker 的参数不需要有任务，Celery 只需要 celery 对象，因此例如，您可以创建一个单独的启动脚本，例如调用 celery_worker.py create_app 创建app 和cel 然后将其作为-A celery_worker.cel 交给worker，完全不涉及蓝图。

希望这会有所帮助。

【讨论】：

@Miguel 这是否意味着我正在创建两个 Flask 实例，一个在 celery_worker.py 中，另一个在 manage.py 中？我也有同样的问题。
@ShulhiSapli 是的，这是两个不同的进程，每个都有自己的应用程序实例。但是两者都应该以相同的方式创建，因此它们实际上是等效的（例如，它们具有相同的配置）。 Celery worker 中应用实例的唯一目的是为需要它的代码提供上下文。您不能以这种方式将session、g、request 变量从一个进程传递到另一个进程。

【解决方案2】：

我解决这个错误的方法是创建两个 Flask 实例，一个用于 Web 应用程序，另一个用于初始 Celery 实例。

就像@Miguel 说的，我有

celery_app.py 用于 celery 实例
Flask 实例的manager.py

在这两个文件中，每个模块都有自己的 Flask 实例。

所以我可以在视图中使用 celery.task。我可以单独启动celery worker。

【讨论】：

【解决方案3】：

感谢Bob Jordan，您可以在https://stackoverflow.com/a/50665633/2794539找到答案，

要点：
1.make_celery同时做两件事：创建celery app和运行带有flask内容的celery，所以你可以创建两个函数来做make_celery工作
2. celery 应用必须在蓝图注册前初始化

【讨论】：

【解决方案4】：

遇到同样的问题，我最终使用shared_task (docs) 非常轻松地解决了它，保留了一个 app.py 文件，而不必多次实例化烧瓶应用程序。

导致循环导入的原始情况：

from src.app import celery  # src.app is ALSO importing the blueprints which are importing this file which causes the circular import.


@celery.task(bind=True)
def celery_test(self):
    sleep(5)
    logger.info("Task processed by Celery.")

当前运行良好且避免循环导入的代码：

# from src.app import celery <- not needed anymore!


@shared_task(bind=True)
def celery_test(self):
    sleep(5)
    logger.info("Task processed by Celery.")

请注意，我对 Celery 还很陌生，所以我可能会监督重要的事情，如果有经验的人可以发表意见，那就太好了。

【讨论】：