Flask - 不能在 app/__init__.py 中使用实例 celery？答案

【问题标题】：Flask - Cannot use instance celery into app/__init__.py?Flask - 不能在 app/__init__.py 中使用实例 celery？
【发布时间】：2020-01-31 09:35:18
【问题描述】：

我已经访问了社区中的几个问题，但我认为这是我的错，因为我无法确定一个合适的方法，在使用documentation Flask 中指出的工厂架构模式。

我们有一个基于 Flask + Gunicorn + Nginx 的应用程序，它始终可以正常工作，没有遇到重大挑战。但是，我们需要构建一个需要异步任务的解决方案。很快，人们一致同意使用 Celery，但即使遵循使用 here 的风格。

我无法编译项目，因为它通知存在导入错误，并且官方documentation of the framework 并没有明确说明如何使用它，在结构工厂中，只是如何在单个实例中执行。

我该如何解决这个问题？

我的回溯：

  File "/home/user/Workspace/test/project_async/celery_runner.py", line 3, in <module>
    from app import create_app
  File "/home/user/Workspace/test/project_async/app/__init__.py", line 9, in <module>
    from .api import configure as config_api
  File "/home/user/Workspace/test/project_async/app/api.py", line 3, in <module>
    from .resources.chosen import Change
  File "/home/user/Workspace/test/project_async/app/resources/chosen.py", line 5, in <module>
    from app.jobs.tasks import exec_change as tk_change
  File "/home/user/Workspace/test/project_async/app/jobs/tasks.py", line 1, in <module>
    from app import celery
ImportError: cannot import name 'celery' from 'app' (/home/user/Workspace/test/project_async/app/__init__.py)

我的结构：

.
├── app
│   ├── api.py
│   ├── jobs
│   │   ├── celeryconfig.py
│   │   ├── __init__.py
│   │   └── tasks.py
│   ├── jwt.py
│   ├── models
│   │   ├── core.py
│   │   ├── __init__.py
│   │   ├── inventory.py
│   │   └── other.py
│   ├── resources
│   │   ├── chosen.py
│   │   └── __init__.py
│   ├── schemas.py
│   ├── services
│   │   ├── chosen.py
│   │   ├── base.py
│   │   ├── chosen_other.py
│   │   └── inventory.py
│   ├── static
│   │   ├── swagger.json
│   │   └── swagger.yaml
│   ├── swagger.py
│   └── utils.py
├── celery_runner.py
├── config.py
├── main.py
├── test.log
└── tests
    ├── __init__.py
    └── test_inventory.py

代码：

路径 - app/__init__.py

from flask import Flask
from flask_cors import CORS
from flask_migrate import Migrate

from config import configure as config_project

from .api import configure as config_api
from .jwt import configure as config_jwt
from .models.core import configure as config_db
from .models.inventory import configure as config_db_mongo
from .swagger import configure as config_docs
from celery import Celery

celery = Celery(__name__)

def create_app(config_name):

    app = Flask(__name__)

    '''Added Configurations'''
    config_project(app)
    config_jwt(app)
    config_api(app)
    config_docs(app)
    config_db(app)
    config_db_mongo(app)
    '''Added Thirds'''
    CORS(app)
    Migrate(app, app.db)
    '''Background Tasks'''
    celery.conf.update(app.config)
    celery.config_from_object('app.jobs.celeryconfig')

    return app

路径 - project_async/celery_worker.py

from dotenv import find_dotenv, load_dotenv

from app import create_app

load_dotenv(find_dotenv())

app = create_app(getenv('FLASK_ENV') or 'default')
app.app_context().push()

路径 - app/jobs/celeryconfig.py

broker_url = 'redis://localhost:6379/1'
result_backend = 'redis://localhost:6379/0'

# import
imports = 'app.jobs.tasks'

# Timezone
timezone = 'America/Bogota'
enable_utc = True

路径 - app/jobs/tasks.py

from app import celery
from app.services.chosen_other import ChangeOpportunity


@celery.task(name='tasks.change')
def exec_change(data, change, client):
    change = ChangeOpportunity(data, change, client)
    change.execute_change()

【问题讨论】：

在我看来你只是有循环导入。一个模块正在导入另一个模块，而另一个模块正在导入另一个模块，而另一个模块最终将再次导入第一个模块。
没错@deceze，我已经发现了这一点，错误在于我在模块app /jobs/tasks.py内调用app/__ init__.py中的芹菜实例的方式。但是如何解决呢？
主要是不要将很多东西放入__init__.py。将所有内容放入显式子模块并显式地从这些子模块中导入。
但是@deceze，在几种风格的项目中，使用flask，我注意到注入外部配置（例如数据库的上下文等）是有效的。显然只用芹菜这似乎不是正确的方法。

标签： python flask design-patterns celery

【解决方案1】：

这是由循环导入引起的。实际问题是导入 api.py 最终会导入您的 Celery 任务。由于您在定义 celery 实例之前执行了 from .api import configure，这与循环依赖关系中断。

您应该尝试重新组织您的代码，以便在导入一个模块时不需要导入大量其他模块。如果您查看堆栈跟踪，您可以看到 api.py 导入 resources/chosen.py，后者导入 jobs/tasks.py。是否需要全部跨包导入？

尝试分解更复杂的模块，以免所有这些间接导入。

【讨论】：