Flask-Mail - Sending email asynchronously, based on Flask-Cookiecutter

Okay, i found solution for my question i posting it here for others developers:

I create file: email.py with code:

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

my view.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."

And when i call http://localhost:5000/mailer it starts countdown and after few seconds is mail sent.

You can move app = Flask(__name__) out of the application factory and place it at the module level. This allows you to pass the app instance with it's application context into your thread for sending the email. You'll likely need to change some imports in other areas to prevent circular dependencies, but it shouldn't be too bad.

Here is an example of how you can do this using Flask-Mail and Flask-RESTful. It also shows how to use pytest for testing this.

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)

And in your module for sending emails you would have something like this:

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 comments)

Note: I posted this years ago and I feel instantiating the flask object outside of the application factory is not ideal. The send_email function will need a flask instance to work, but you can instantiate a new flask app in the function (don't forget your config).

I would guess that current_app may also work but I feel that might have side effects given that it would need to create a new app context with-in a current app context, seems wrong, but might work.

A good option to consider: Look into celery and use RabbitMQ for your backend.
For larger apps or volumes you may look into decoupling the mailing of emails into a different app via message broker like RabbitMQ. Look into Event Driven Design patterns. This may be attractive if you have multiple apps that will need a mailing service. This could be nice if your service needs to support audit logs, delivery recovery, etc.

Move email send function to a background thread:

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