使用Flask+RQ发送确认邮件时遇应用上下文外运行错误

问题描述 #

新用户注册时直接调用发送确认邮件功能正常，但将该任务放到RQ后台队列运行时出现错误，初始错误指向msg = Message()行。此前其他后台任务运行无异常。

初始错误栈 #

File "/Users/martifont/Dev/bz/./FlaskApp/utils.py", line 22, in send_confirmation_email
    msg = Message()

相关代码 #

users.py（注册路由） #

#imports
from .utils import send_reset_email, send_confirmation_email, example

users = Blueprint('users', __name__)

#SIGNUP
@users.route('/signup', methods=['GET', 'POST'])
def signup():
form = SignupForm()
if form.validate_on_submit():
user = User(email=form.email.data, username=form.username.data, confirmed=False, is_admin=False, password = generate_password_hash(form.password.data, method='sha256'))
db.session.add(user)
db.session.commit()

        user_id=user.id
        with app.app_context():
            job = q.enqueue(send_confirmation_email, user)
    
        login_user(user)
        flash('A confirmation email has been sent to your inbox.', 'is-success')
        return redirect(url_for('users.account', id=user.id))
    return render_template('signup.html', form=form)

utils.py（邮件发送函数） #

def send_confirmation_email(user):
    token = user.get_confirmation_token()
    msg = Message()
    msg.subject = ('Email Confirmation')
    msg.sender = 'MAIL_DEFAULT_SENDER'
    msg.recipients = [user.email]
    msg.body = f'''
Welcome { user.username }!,
Thanks for signing up. Please follow this link to activate your account:
{url_for('users.confirm_email', token=token, _external=True)}
Thanks!

'''
    mail.send(msg)

编辑补充：调整参数后的新错误 #

尝试传入app和user作为参数后，出现序列化相关错误：

File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/flask/app.py", line 2548, in __call__
    return self.wsgi_app(environ, start_response)
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/flask/app.py", line 2528, in wsgi_app
    response = self.handle_exception(e)
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/flask/app.py", line 2525, in wsgi_app
    response = self.full_dispatch_request()
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/flask/app.py", line 1822, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/flask/app.py", line 1820, in full_dispatch_request
    rv = self.dispatch_request()
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/flask/app.py", line 1796, in dispatch_request
    return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)
  File "/Users/martifont/Dev/bz/FlaskApp/users.py", line 30, in signup
    job = q.enqueue(send_confirmation_email, args=(app, user))
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/rq/queue.py", line 514, in enqueue
    return self.enqueue_call(
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/rq/queue.py", line 410, in enqueue_call
    return self.enqueue_job(job, pipeline=pipeline, at_front=at_front)
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/rq/queue.py", line 572, in enqueue_job
    job.save(pipeline=pipe)
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/rq/job.py", line 694, in save
    mapping = self.to_dict(include_meta=include_meta)
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/rq/job.py", line 629, in to_dict
    'data': zlib.compress(self.data),
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/rq/job.py", line 305, in data
    self._data = self.serializer.dumps(job_tuple)

解决方案 #

1. 修复后台任务的Flask上下文缺失 #

后台任务运行在独立进程，没有默认Flask上下文，Message、url_for、mail.send均依赖上下文，需在任务内部手动创建：
修改utils.py：

from flask import current_app, url_for
from flask_mail import Message
from . import mail

def send_confirmation_email(user_id):
    # 延迟导入模型避免循环依赖
    from .models import User
    user = User.query.get(user_id)
    if not user:
        return
    
    token = user.get_confirmation_token()
    # 从配置中获取真实发件人邮箱
    sender = current_app.config['MAIL_DEFAULT_SENDER']
    
    with current_app.app_context():
        msg = Message('邮箱确认', sender=sender, recipients=[user.email])
        msg.body = f'''
欢迎 { user.username }！
感谢注册，请点击以下链接激活你的账号：
{url_for('users.confirm_email', token=token, _external=True)}
谢谢！
'''
        mail.send(msg)

2. 避免传递不可序列化对象 #

RQ需要序列化任务参数，直接传递User或app实例会失败，改用传递user_id，在任务内部查询用户：
修改users.py中的任务调用：

# 移除原有的with app.app_context()包裹，上下文在任务内部处理
job = q.enqueue(send_confirmation_email, user.id)

3. 确保RQ Worker加载Flask上下文 #

创建worker.py文件，让Worker启动时加载Flask应用配置和上下文：

from flask import Flask
from rq import Worker, Queue, Connection
import redis
from FlaskApp import mail
from FlaskApp.models import User

# 初始化Flask应用，与项目初始化逻辑一致
app = Flask(__name__)
app.config.from_object('config')

# 初始化Redis连接
redis_conn = redis.from_url(app.config['REDIS_URL'])
q = Queue(connection=redis_conn)

if __name__ == '__main__':
    with app.app_context():
        worker = Worker([q], connection=redis_conn)
        worker.work()

启动Worker时执行python worker.py，而非直接用rq worker。

4. 修正发件人配置 #

确保MAIL_DEFAULT_SENDER在配置文件中已正确设置为真实邮箱地址，而非字符串'MAIL_DEFAULT_SENDER'。

内容的提问来源于stack exchange，提问作者Marti F.R