首页 文章

无法从烧瓶应用程序导入Celery对象

提问于
浏览 次
1

我正在尝试学习并将Celery与我的烧瓶应用程序集成 . 我尝试了链接中提到的事情：Flask Celery

但我想在单独的模块中移动所有异步任务 . 所以这是我的git存储库，我正在进行更改：Beacons Repo

所以我在_1716376中初始化我的芹菜对象：

from beacons.async import make_celery


mq_url = 'amqp://mq:mq@localhost/localhost'


app = Flask(__name__)

app.config.update(
    CELERY_BROKER_URL=mq_url,
    CELERY_BACKEND=mq_url
)

log_handler = logging.StreamHandler()
log_handler.setLevel(logging.DEBUG)

log_handler.setFormatter(Formatter(
    '\n-------------------------------------\n'
    'TIME: %(asctime)s \n'
    'LEVEL: %(levelname)s \n'
    'MESSAGE: %(message)s \n'
    'FILE: %(pathname)s \n'
    'LINE: %(lineno)d'
    '\n-------------------------------------\n'))

app.logger.addHandler(log_handler)

app.register_blueprint(portal, url_prefix='/beacons')
app.register_blueprint(apis, url_prefix='/beacons/apis')
celery = make_celery(app)


__all__ = ['celery']

我的 make_celery 在 beacons/async/__init__.py 中如下：

from celery import Celery


def make_celery(app):
    celery = Celery(app.import_name, backend=app.config['CELERY_BACKEND'],
                    broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)
    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

我创建了另一个模块 beacons/tasks.py ，它是：

from beacons import celery


@celery.task()
def add(x, y):
    return x + y

我在我的一个观点 beacons/portal/view/rest_apis.py 中使用此模块如下：

from beacons.tasks import add

@apis.route('/add', methods=['POST'])
def add_page():
    x = request.form.get('x')
    y = request.form.get('y')

    task = add.delay(x, y)
    return json_response(
        {"id": task.id}, status_code=200)

当我尝试运行Celery worker和Flask App时，我收到此错误：

from beacons import celery
ImportError: cannot import name celery

如何将 beacons 模块中编写的芹菜对象导入其子模块或任务本身 . 有关完整代码，请参阅Git Repo .

python flask celery

1 回答

  • 0

    我尝试运行您的代码，不要收到此错误 . 我在你的run.py中注释掉你的系统参数，并在下面留下这段代码 .

    if __name__ == '__main__':
    app.secret_key = str(uuid.uuid4())
    port = '9000'
    app.run(debug=False, port=port, host='0.0.0.0')

    然后我执行python run.py enter image description here

    然后我请求127.0.0.1:9000烧瓶并获得enter image description here

    Flask工作正常，除了日志格式错误，并且此错误超出了主题 .

    你能发布你执行的命令和你执行的目录吗？我认为您的工作目录可能会导致此ImportError .

    PS：你在GitHub中给出的分支代码与此问题中的代码帖不同 .

    回复于

相关问题