更新Flask-Security-Too后，重构异步邮件发送（send_mail_task已弃用）

新版Flask-Security-Too已移除@security.send_mail_task装饰器，改用配置项指定异步任务函数的方式实现邮件异步发送，以下是具体重构方案：

核心思路 #

• 不再通过装饰器注册异步邮件任务，而是通过配置SECURITY_SEND_MAIL_TASK直接指向负责异步发送的函数
• 异步函数需自行处理应用上下文（因为异步任务脱离请求上下文）

代码示例 #

1. 基础线程异步实现 #

适合小型项目，用Python线程实现简单异步：

from flask_mail import Mail
from flask import Flask
import threading

mail = Mail()

def create_app(test_config=None):
    app = Flask(__name__)
    
    # 配置邮件服务（根据你的实际服务商调整）
    app.config.update(
        MAIL_SERVER='smtp.example.com',
        MAIL_PORT=587,
        MAIL_USE_TLS=True,
        MAIL_USERNAME='your-email@example.com',
        MAIL_PASSWORD='your-password'
    )
    
    # 关键配置：指定异步邮件任务函数的可导入路径
    app.config['SECURITY_SEND_MAIL_TASK'] = 'app.delay_security_email'
    
    # 初始化扩展
    mail.init_app(app)
    
    # 初始化Flask-Security-Too（替换为你的用户/角色模型逻辑）
    from flask_security import Security, SQLAlchemyUserDatastore, User, Role
    from flask_sqlalchemy import SQLAlchemy
    db = SQLAlchemy(app)
    user_datastore = SQLAlchemyUserDatastore(db, User, Role)
    Security(app, user_datastore)
    
    return app

# 异步邮件任务函数，需放在可被全局导入的位置
def delay_security_email(msg):
    def send_in_thread():
        # 手动创建应用上下文，确保Mail扩展能正常工作
        from app import app
        with app.app_context():
            mail.send(msg)
    # 启动线程执行发送
    threading.Thread(target=send_in_thread).start()

2. Celery任务队列实现 #

如果项目使用Celery处理异步任务，可按以下方式重构：

from flask_mail import Mail
from flask import Flask
from celery import Celery

mail = Mail()
# 初始化Celery（根据你的消息中间件调整，比如Redis/RabbitMQ）
celery = Celery('app', broker='redis://localhost:6379/0')

def create_app(test_config=None):
    app = Flask(__name__)
    
    # 邮件及Security配置
    app.config.update(
        MAIL_SERVER='smtp.example.com',
        # ...其他邮件配置
        SECURITY_SEND_MAIL_TASK = 'app.delay_security_email'
    )
    
    mail.init_app(app)
    
    # 初始化Flask-Security-Too...
    from flask_security import Security, SQLAlchemyUserDatastore, User, Role
    from flask_sqlalchemy import SQLAlchemy
    db = SQLAlchemy(app)
    user_datastore = SQLAlchemyUserDatastore(db, User, Role)
    Security(app, user_datastore)
    
    return app

# 用Celery装饰器标记异步任务
@celery.task
def delay_security_email(msg):
    from app import app, mail
    with app.app_context():
        mail.send(msg)

注意事项 #

• SECURITY_SEND_MAIL_TASK的值必须是可全局导入的函数路径，格式为'包名.模块名.函数名'
• 异步函数内部必须手动创建app.app_context()，否则Flask-Mail无法获取应用配置
• 如果使用Celery，需确保Celery实例能正确加载应用上下文（可通过celery.conf.update(app.config)同步配置）

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