Flask集成Celery报错：'str' object has no attribute 'level' 求助

问题根源 #

你在celery_init_app函数中错误使用了config_from_object方法加载Celery配置——该方法用于加载Python模块/类对象的配置，而你传入的是字典格式的app.config['CELERY']，导致Celery内部处理配置时触发类型错误。

修复步骤 #

1. 替换配置加载方法
   将celery_app.config_from_object(app.config['CELERY'])改为celery_app.config_from_dict(app.config['CELERY'])，config_from_dict是Celery专门用于加载字典格式配置的方法。

2. 使用正确的Celery启动命令
   启动Worker时必须明确指定Celery应用实例的路径，命令格式如下：

   celery -A make_celery:celery_app worker --loglevel=info
   

   其中make_celery是你的Python文件名，celery_app是文件中导出的Celery实例变量。

修改后的完整代码 #

app/init.py #

from flask import Flask
from celery import Celery, Task

def celery_init_app(app: Flask):
    class FlaskTask(Task):
        def __call__(self, *args: object, **kwargs: object):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery_app = Celery(app.name, task_cls=FlaskTask)
    # 替换为config_from_dict加载字典配置
    celery_app.config_from_dict(app.config['CELERY'])
    celery_app.set_default()
    app.extensions['celery'] = celery_app 
    return celery_app 

def create_app(config_class=Config):
    app = Flask(__name__)
    app.config.from_object(config_class)
    app.config.from_mapping(
        CELERY=dict(
            broker_url="pyamqp://guest@localhost//",
            result_backend="redis://localhost",
            task_serializer='json',
            result_serializer='json',
            accept_content=['json'],
            task_ignore_result=True,
        ),
    )
    app.config.from_prefixed_env()

    # ... 其他配置代码
    celery_init_app(app)
    return app

make_celery.py #

from app import create_app 

flask_app = create_app()
celery_app = flask_app.extensions["celery"]

额外检查点 #

• 确认已安装依赖：celery、redis（若用Redis做结果后端）、amqp（若用RabbitMQ做Broker）。
• 确保RabbitMQ和Redis服务在本地正常启动运行。

内容的提问来源于stack exchange，提问作者JTPing