Recuerda que en la informática cualquier problema casi siempre tiene diversas resoluciones, así que enseñamos lo mejor y más óptimo.
Solución:
De acuerdo, encontré una solución para mi pregunta y la publico aquí para otros desarrolladores:
Creo el archivo: email.py con el código:
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
mi vista.py:
...
from app.email import send_email
...
@blueprint.route('/mailer', methods=['GET', 'POST'])
def mailer():
user = current_user.full_name
send_email(('[email protected]'),
'New mail', 'test.html',
user=user)
return "Mail has been send."
Y cuando llamo a http://localhost:5000/mailer, comienza la cuenta regresiva y después de unos segundos se envía el correo.
Puedes moverte app = Flask(__name__) fuera de la fábrica de aplicaciones y colóquelo en el nivel del módulo. Esto le permite pasar la instancia de la aplicación con su contexto de aplicación a su hilo para enviar el correo electrónico. Es probable que deba cambiar algunas importaciones en otras áreas para evitar dependencias circulares, pero no debería ser tan malo.
Aquí hay un ejemplo de cómo puede hacer esto usando Flask-Mail y Flask-RESTful. También muestra cómo usar pytest para probar esto.
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)
Y en tu módulo de envío de emails tendrías algo como esto:
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)
(comentarios de 2019)
Nota: publiqué esto hace años y creo que crear una instancia del objeto matraz fuera de la fábrica de aplicaciones no es lo ideal. los send_email La función necesitará una instancia de matraz para funcionar, pero puede crear una instancia de una nueva aplicación de matraz en la función (no olvide su configuración).
supongo que current_app también puede funcionar, pero creo que podría tener efectos secundarios dado que necesitaría crear un nuevo contexto de aplicación dentro de un contexto de aplicación actual, parece incorrecto, pero podría funcionar.
Una buena opción a considerar: busque apio y use RabbitMQ para su backend.
Para aplicaciones o volúmenes más grandes, puede considerar desacoplar el envío de correos electrónicos en una aplicación diferente a través de un agente de mensajes como RabbitMQ. Mire los patrones de diseño impulsado por eventos. Esto puede ser atractivo si tiene varias aplicaciones que necesitarán un servicio de correo. Esto podría ser bueno si su servicio necesita admitir registros de auditoría, recuperación de entrega, etc.
Mueva la función de envío de correo electrónico a un hilo de fondo:
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
Si posees alguna perplejidad y disposición de aclarar nuestro división puedes añadir un informe y con deseo lo analizaremos.