发送具有正确上下文的电子邮件任务答案

【问题标题】：Send email task with correct context发送具有正确上下文的电子邮件任务
【发布时间】：2026-02-18 20:30:02
【问题描述】：

这段代码是我的 celery worker 脚本：

from app import celery, create_app
app = create_app('default')
app.app_context().push()

当我尝试运行 worker 时，我会遇到这个错误：

File "/home/vagrant/myproject/venv/app/mymail.py", line 29, in send_email_celery
    msg.html = render_template(template + '.html', **kwargs)
  File "/home/vagrant/myproject/venv/local/lib/python2.7/site-packages/flask/templating.py", line 126, in render_template
    ctx.app.update_template_context(context)
  File "/home/vagrant/myproject/venv/local/lib/python2.7/site-packages/flask/app.py", line 716, in update_template_context
    context.update(func())
TypeError: 'NoneType' object is not iterable

我的问题是如何在 celery 中使用工人时发送电子邮件任务。

mymail.py

from flask import current_app, render_template
from flask.ext.mail import Message
from . import mail, celery

@celery.task
def send_async_email_celery(msg):
    mail.send(msg)

def send_email_celery(to, subject, template, **kwargs):
    app = current_app._get_current_object()
    msg = Message(subject, sender=app.config['MAIL_SENDER'], recipients=[to])
    msg.html = render_template(template + '.html', **kwargs)
    send_async_email_celery.delay(msg)

__init__

...

def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)

    bootstrap.init_app(app)
    mail.init_app(app)
    db.init_app(app)
    login_manager.init_app(app)
    celery.conf.update(app.config)

    redis_store.init_app(app)

    from .users import main as main_blueprint
    app.register_blueprint(main_blueprint)

    return app

显然蓝图和工人之间存在一些冲突。如果可能的话，删除蓝图不是一个选项，因为我需要在电子邮件模板中使用自定义过滤器。

【问题讨论】：

我不遵循这里的推理。你想让芹菜工人注册蓝图吗？对我来说听起来倒退了，我希望有一个蓝图来注册 celery 任务，而不是相反。你能解释一下蓝图应该做什么以及芹菜工人应该做什么吗？
另外，你的模板里有什么？
@user3012759 即使使用空模板（仅用于调试），我也会收到错误消息。所以问题不在于模板。

标签： python python-2.7 flask celery celery-task

【解决方案1】：

用这个code调试后终于找到问题的原因了。

我有一个不会返回任何结果的app_context_processor。

@mod.app_context_processor
def last_reputation_changes():
    if current_user:
        #code
        return dict(reputation='xxx')

发送电子邮件时，current_user 需要 else 大小写才能返回，因为来自 from flask.ext.login import current_user 的 current_user 未定义。基本上我只需要这样的东西。

def last_reputation_changes():
    if current_user:
        #code
        return dict(reputation='xxx')
    else:
        return dict(reputation=None)

所以问题与 celery 无关，而与烧瓶登录集成有关。

【讨论】：