Flask-Mail - 基于 Flask-Cookiecutter 异步发送电子邮件
Flask-Mail - Sending email asynchronously, based on Flask-Cookiecutter
我的flask项目基于Flask-Cookiecutter，需要异步发送邮件
发送邮件的功能由Miguel’s Tutorial配置，同步发送可以正常工作，但我不知道如何修改它以异步发送。
我的app.py
def create_app(config_object=ProdConfig):
app = Flask(__name__)
app.config.from_object(config_object)
register_extensions(app)
register_blueprints(app)
register_errorhandlers(app)
return app
def register_extensions(app):
assets.init_app(app)
bcrypt.init_app(app)
cache.init_app(app)
db.init_app(app)
login_manager.init_app(app)
debug_toolbar.init_app(app)
migrate.init_app(app, db)
mail.init_app(app)
return None
我的view.py
from flask import current_app
@async
def send_async_email(current_app, msg):
with current_app.app_context():
print('##### spustam async')
mail.send(msg)
# Function for sending emails
def send_email(to, subject, template, **kwargs):
msg = Message(subject, recipients=[to])
msg.html = render_template('emails/' + template, **kwargs)
send_async_email(current_app, msg)
路线在view.py
@blueprint.route('/mailer', methods=['GET', 'POST'])
def mailer():
user = current_user.full_name
send_email(('name@gmail.com'),
'New mail', 'test.html',
user=user)
return "Mail has been send."
Application 运行 在我的本地主机中，它以命令开头：
python manage.py server
当我调用发送邮件的函数时，控制台输出为：
RuntimeError: Working outside of application context.
This typically means that you attempted to use functionality that needed
to interface with the current application object in a way. To solve
this set up an application context with app.app_context(). See the
documentation for more information.
感谢您的回答。
将电子邮件发送功能移动到后台线程：
from threading import Thread
def send_async_email(app,msg):
with current_app.app_context():
mail.send(msg)
def send_email(to, subject, template, **kwargs):
msg = Message(subject, recipients=[to])
msg.html = render_template('emails/' + template, **kwargs)
thr = Thread(target=send_async_email,args=[app,msg])
thr.start()
return thr
好的，我找到了我的问题的解决方案，我将它发布在这里供其他开发人员使用：
我创建文件：email.py，代码：
from threading import Thread
from flask import current_app, render_template
from flask_mail import Message
from .extensions import mail
from time import sleep
def send_async_email(app, msg):
with app.app_context():
# block only for testing parallel thread
for i in range(10, -1, -1):
sleep(2)
print('time:', i)
print('====> sending async')
mail.send(msg)
def send_email(to, subject, template, **kwargs):
app = current_app._get_current_object()
msg = Message(subject, recipients=[to])
msg.html = render_template('emails/' + template, **kwargs)
thr = Thread(target=send_async_email, args=[app, msg])
thr.start()
return thr
我的view.py:
...
from app.email import send_email
...
@blueprint.route('/mailer', methods=['GET', 'POST'])
def mailer():
user = current_user.full_name
send_email(('name@gmail.com'),
'New mail', 'test.html',
user=user)
return "Mail has been send."
当我调用 http://localhost:5000/mailer 时，它开始倒计时，几秒钟后邮件发送。
您可以将app = Flask(__name__)移出应用程序工厂并将其放置在模块级别。这允许您将应用程序实例及其应用程序上下文传递到您的线程中以发送电子邮件。您可能需要更改其他区域的一些导入以防止循环依赖，但这应该不会太糟糕。
这是使用 Flask-Mail 和 Flask-RESTful 的 example of how you can do this。它还展示了如何使用 pytest 对此进行测试。
from flask import Flask
from .extensions import mail
from .endpoints import register_endpoints
from .settings import ProdConfig
# app context needs to be accessible at the module level
# for the send_message.send_
app = Flask(__name__)
def create_app(config=ProdConfig):
""" configures and returns the the flask app """
app.config.from_object(config)
register_extensions()
register_endpoints(app)
return app
def register_extensions():
""" connects flask extensions to the app """
mail.init_app(app)
在你发送电子邮件的模块中，你会有这样的东西：
from flask_mail import Message
from app import app
from app import mail
from utils.decorators import async_task
def send_email(subject, sender, recipients, text_body, html_body=None, **kwargs):
app.logger.info("send_email(subject='{subject}', recipients=['{recp}'], text_body='{txt}')".format(sender=sender, subject=subject, recp=recipients, txt=text_body))
msg = Message(subject, sender=sender, recipients=recipients, **kwargs)
msg.body = text_body
msg.html = html_body
app.logger.info("Message(to=[{m.recipients}], from='{m.sender}')".format(m=msg))
_send_async_email(app, msg)
@async_task
def _send_async_email(flask_app, msg):
""" Sends an send_email asynchronously
Args:
flask_app (flask.Flask): Current flask instance
msg (Message): Message to send
Returns:
None
"""
with flask_app.app_context():
mail.send(msg)
（2019 条评论）
注意：我在几年前发帖，我觉得在 application factory 之外实例化烧瓶对象并不理想。 send_email 函数需要一个 flask 实例才能工作，但你可以在函数中实例化一个新的 flask 应用程序（不要忘记你的配置）。
我猜想 current_app 也可以工作，但我觉得这可能会产生副作用，因为它需要创建一个新的应用程序上下文 with-in 当前应用程序上下文，这似乎是错误的，但是可能有用。
一个值得考虑的好选择：查看 celery and use RabbitMQ 的后端。
对于较大的应用程序或容量较大的应用程序，您可以考虑通过 RabbitMQ 等消息代理将电子邮件的邮寄解耦到不同的应用程序。查看事件驱动设计模式。如果您有多个需要邮寄服务的应用程序，这可能很有吸引力。如果您的服务需要支持审核日志、交付恢复等，这可能很好。
我的flask项目基于Flask-Cookiecutter，需要异步发送邮件
发送邮件的功能由Miguel’s Tutorial配置，同步发送可以正常工作，但我不知道如何修改它以异步发送。
我的app.py
def create_app(config_object=ProdConfig):
app = Flask(__name__)
app.config.from_object(config_object)
register_extensions(app)
register_blueprints(app)
register_errorhandlers(app)
return app
def register_extensions(app):
assets.init_app(app)
bcrypt.init_app(app)
cache.init_app(app)
db.init_app(app)
login_manager.init_app(app)
debug_toolbar.init_app(app)
migrate.init_app(app, db)
mail.init_app(app)
return None
我的view.py
from flask import current_app
@async
def send_async_email(current_app, msg):
with current_app.app_context():
print('##### spustam async')
mail.send(msg)
# Function for sending emails
def send_email(to, subject, template, **kwargs):
msg = Message(subject, recipients=[to])
msg.html = render_template('emails/' + template, **kwargs)
send_async_email(current_app, msg)
路线在view.py
@blueprint.route('/mailer', methods=['GET', 'POST'])
def mailer():
user = current_user.full_name
send_email(('name@gmail.com'),
'New mail', 'test.html',
user=user)
return "Mail has been send."
Application 运行 在我的本地主机中，它以命令开头：
python manage.py server
当我调用发送邮件的函数时，控制台输出为：
RuntimeError: Working outside of application context.
This typically means that you attempted to use functionality that needed
to interface with the current application object in a way. To solve
this set up an application context with app.app_context(). See the
documentation for more information.
感谢您的回答。
将电子邮件发送功能移动到后台线程：
from threading import Thread
def send_async_email(app,msg):
with current_app.app_context():
mail.send(msg)
def send_email(to, subject, template, **kwargs):
msg = Message(subject, recipients=[to])
msg.html = render_template('emails/' + template, **kwargs)
thr = Thread(target=send_async_email,args=[app,msg])
thr.start()
return thr
好的，我找到了我的问题的解决方案，我将它发布在这里供其他开发人员使用：
我创建文件：email.py，代码：
from threading import Thread
from flask import current_app, render_template
from flask_mail import Message
from .extensions import mail
from time import sleep
def send_async_email(app, msg):
with app.app_context():
# block only for testing parallel thread
for i in range(10, -1, -1):
sleep(2)
print('time:', i)
print('====> sending async')
mail.send(msg)
def send_email(to, subject, template, **kwargs):
app = current_app._get_current_object()
msg = Message(subject, recipients=[to])
msg.html = render_template('emails/' + template, **kwargs)
thr = Thread(target=send_async_email, args=[app, msg])
thr.start()
return thr
我的view.py:
...
from app.email import send_email
...
@blueprint.route('/mailer', methods=['GET', 'POST'])
def mailer():
user = current_user.full_name
send_email(('name@gmail.com'),
'New mail', 'test.html',
user=user)
return "Mail has been send."
当我调用 http://localhost:5000/mailer 时，它开始倒计时，几秒钟后邮件发送。
您可以将app = Flask(__name__)移出应用程序工厂并将其放置在模块级别。这允许您将应用程序实例及其应用程序上下文传递到您的线程中以发送电子邮件。您可能需要更改其他区域的一些导入以防止循环依赖，但这应该不会太糟糕。
这是使用 Flask-Mail 和 Flask-RESTful 的 example of how you can do this。它还展示了如何使用 pytest 对此进行测试。
from flask import Flask
from .extensions import mail
from .endpoints import register_endpoints
from .settings import ProdConfig
# app context needs to be accessible at the module level
# for the send_message.send_
app = Flask(__name__)
def create_app(config=ProdConfig):
""" configures and returns the the flask app """
app.config.from_object(config)
register_extensions()
register_endpoints(app)
return app
def register_extensions():
""" connects flask extensions to the app """
mail.init_app(app)
在你发送电子邮件的模块中，你会有这样的东西：
from flask_mail import Message
from app import app
from app import mail
from utils.decorators import async_task
def send_email(subject, sender, recipients, text_body, html_body=None, **kwargs):
app.logger.info("send_email(subject='{subject}', recipients=['{recp}'], text_body='{txt}')".format(sender=sender, subject=subject, recp=recipients, txt=text_body))
msg = Message(subject, sender=sender, recipients=recipients, **kwargs)
msg.body = text_body
msg.html = html_body
app.logger.info("Message(to=[{m.recipients}], from='{m.sender}')".format(m=msg))
_send_async_email(app, msg)
@async_task
def _send_async_email(flask_app, msg):
""" Sends an send_email asynchronously
Args:
flask_app (flask.Flask): Current flask instance
msg (Message): Message to send
Returns:
None
"""
with flask_app.app_context():
mail.send(msg)
（2019 条评论）
注意：我在几年前发帖，我觉得在 application factory 之外实例化烧瓶对象并不理想。 send_email 函数需要一个 flask 实例才能工作，但你可以在函数中实例化一个新的 flask 应用程序（不要忘记你的配置）。
我猜想 current_app 也可以工作，但我觉得这可能会产生副作用，因为它需要创建一个新的应用程序上下文 with-in 当前应用程序上下文，这似乎是错误的，但是可能有用。
一个值得考虑的好选择：查看 celery and use RabbitMQ 的后端。
对于较大的应用程序或容量较大的应用程序，您可以考虑通过 RabbitMQ 等消息代理将电子邮件的邮寄解耦到不同的应用程序。查看事件驱动设计模式。如果您有多个需要邮寄服务的应用程序，这可能很有吸引力。如果您的服务需要支持审核日志、交付恢复等，这可能很好。