Flask+Celery任务读取app配置遇上下文错误，求解决方法

问题背景 #

项目采用Flask应用工厂模式，通过celery_utils.py配置Celery实例，需要在Celery任务中读取Flask应用的配置值，但调用current_app.config时触发RuntimeError: Working outside of application context，尝试手动创建上下文又出现UnboundLocalError。

相关代码 #

蓝图路由（admin_blueprint） #

@admin_blueprint.route("/send_notification", methods=["GET"])
def send_notification():
    tasks_telegram.send_telegram_notification.apply_async(
                args=["This is a test message"])

初始Celery任务（task_telegram.py） #

from celery import shared_task
from flask import current_app

from ..core import telegram

@shared_task
def send_telegram_notification(message: str):
    app = current_app
    telegram_bot_key = app.config.get("TELEGRAM_BOT_KEY", "")
    telegram_chat_id = app.config.get("TELEGRAM_CHAT_ID", "")
    telegram_obj = telegram.Telegram(bot_key=telegram_bot_key, chat_id=telegram_chat_id)
    telegram_obj.send_telegram_notification(message=message)

触发的错误信息 #

[2023-08-06 22:59:07,648: ERROR/MainProcess] Task app.tasks.tasks_telegram.send_telegram_notification[efd2ca52-6d0c-433b-8c9b-104255ae3f46] raised unexpected: RuntimeError('Working outside of application context.\n\nThis typically means that you attempted to use functionality that needed\nthe current application. To solve this, set up an application context\nwith app.app_context(). See the documentation for more information.')
...
RuntimeError: Working outside of application context.

This typically means that you attempted to use functionality that needed
the current application. To solve this, set up an application context
with app.app_context(). See the documentation for more information.

错误的改写尝试及对应报错 #

from celery import shared_task
from flask import current_app

from ..core import telegram

@shared_task
def send_telegram_notification(message: str):
    with app.app_context():
        app = current_app
        telegram_bot_key = app.config.get("TELEGRAM_BOT_KEY", "")
        telegram_chat_id = app.config.get("TELEGRAM_CHAT_ID", "")
        telegram_obj = telegram.Telegram(
            bot_key=telegram_bot_key, chat_id=telegram_chat_id
        )
        telegram_obj.send_telegram_notification(message=message)

报错：

line 10, in send_telegram_notification
    with app.app_context():
UnboundLocalError: local variable 'app' referenced before assignment

解决方案 #

Celery worker运行在独立进程，默认没有Flask应用上下文，需通过以下方式处理：

方法1：利用Celery的ContextTask自动注入上下文 #

如果你的celery_utils.py是按教程实现的，应该包含绑定Flask应用的ContextTask逻辑，示例如下：

# celery_utils.py
from celery import Celery

def make_celery(app):
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL']
    )
    celery.conf.update(app.config)
    
    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)
    
    celery.Task = ContextTask
    return celery

此时任务会自动在应用上下文内执行，直接修改任务代码即可：

from celery import shared_task
from flask import current_app

from ..core import telegram

@shared_task
def send_telegram_notification(message: str):
    telegram_bot_key = current_app.config.get("TELEGRAM_BOT_KEY", "")
    telegram_chat_id = current_app.config.get("TELEGRAM_CHAT_ID", "")
    telegram_obj = telegram.Telegram(bot_key=telegram_bot_key, chat_id=telegram_chat_id)
    telegram_obj.send_telegram_notification(message=message)

方法2：手动创建应用实例并推入上下文 #

若未使用ContextTask，可在任务中导入应用工厂函数创建实例，再手动推入上下文：

from celery import shared_task
from .. import create_app  # 导入你的应用工厂函数
from ..core import telegram

@shared_task
def send_telegram_notification(message: str):
    app = create_app()  # 创建应用实例，确保能正确加载配置
    with app.app_context():
        telegram_bot_key = app.config.get("TELEGRAM_BOT_KEY", "")
        telegram_chat_id = app.config.get("TELEGRAM_CHAT_ID", "")
        telegram_obj = telegram.Telegram(bot_key=telegram_bot_key, chat_id=telegram_chat_id)
        telegram_obj.send_telegram_notification(message=message)

方法3：直接传入配置参数到任务 #

避免在任务中依赖Flask应用，调用任务时直接传递配置值：

# 蓝图路由修改
@admin_blueprint.route("/send_notification", methods=["GET"])
def send_notification():
    bot_key = current_app.config.get("TELEGRAM_BOT_KEY", "")
    chat_id = current_app.config.get("TELEGRAM_CHAT_ID", "")
    tasks_telegram.send_telegram_notification.apply_async(
        args=["This is a test message", bot_key, chat_id])

# 任务代码修改
from celery import shared_task
from ..core import telegram

@shared_task
def send_telegram_notification(message: str, bot_key: str, chat_id: str):
    telegram_obj = telegram.Telegram(bot_key=bot_key, chat_id=chat_id)
    telegram_obj.send_telegram_notification(message=message)

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