6

My flask project is based on Flask-Cookiecutter and I need to send emails asynchronously.

Function for sending email was configured by Miguel’s Tutorial and sending synchronously works fine, but i don’t know, how I can modify it for sending asynchronously.

My 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

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

route in view.py

@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."

Application run in my localhost and it's starting with command:

python manage.py server

When i call function for sending mail, output in console is:

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.

Thanks for any answer.

3 Answers 3

14

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.

2
  • 1
    it is the '_get_current_object()' that helped me out.
    – Tyler C.
    Commented Jul 28, 2019 at 4:59
  • 1
    Thanks.. I struggled with it too for bit until we landed on this solution.
    – Sajjad
    Commented Mar 6 at 3:36
4

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
1
  • @shefget-lika: thanks for your answer, but console still show message: Working outside of application context. In the meantime i find out correcnt solution. I will post it here for others developers. Commented Oct 30, 2016 at 12:12
4

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.

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.