Python Flask with celery out of application context

Question

I am building a website using python Flask. Everything is going good and now I am trying to implement celery.

That was going good as well until I tried to send an email using flask-mail from celery. Now I am getting an "working outside of application context" error.

full traceback is

  Traceback (most recent call last):
  File "/usr/lib/python2.7/site-packages/celery/task/trace.py", line 228, in trace_task
    R = retval = fun(*args, **kwargs)
  File "/usr/lib/python2.7/site-packages/celery/task/trace.py", line 415, in __protected_call__
    return self.run(*args, **kwargs)
  File "/home/ryan/www/CG-Website/src/util/mail.py", line 28, in send_forgot_email
    msg = Message("Recover your Crusade Gaming Account")
  File "/usr/lib/python2.7/site-packages/flask_mail.py", line 178, in __init__
    sender = current_app.config.get("DEFAULT_MAIL_SENDER")
  File "/usr/lib/python2.7/site-packages/werkzeug/local.py", line 336, in __getattr__
    return getattr(self._get_current_object(), name)
  File "/usr/lib/python2.7/site-packages/werkzeug/local.py", line 295, in _get_current_object
    return self.__local()
  File "/usr/lib/python2.7/site-packages/flask/globals.py", line 26, in _find_app
    raise RuntimeError('working outside of application context')
RuntimeError: working outside of application context

This is my mail function:

@celery.task
def send_forgot_email(email, ref):
    global mail
    msg = Message("Recover your Crusade Gaming Account")
    msg.recipients = [email]
    msg.sender = "Crusade Gaming [email protected]"
    msg.html = \
        """
        Hello Person,<br/>

        You have requested your password be reset. <a href="{0}" >Click here recover your account</a> or copy and paste this link in to your browser: {0} <br />

        If you did not request that your password be reset, please ignore this.
        """.format(url_for('account.forgot', ref=ref, _external=True))
    mail.send(msg)

This is my celery file:

from __future__ import absolute_import

from celery import Celery

celery = Celery('src.tasks',
                broker='amqp://',
                include=['src.util.mail'])


if __name__ == "__main__":
    celery.start()

Solution 1

Here is a solution which works with the flask application factory pattern and also creates celery task with context, without needing to use app.app_context(). It is really tricky to get that app while avoiding circular imports, but this solves it. This is for celery 4.2 which is the latest at the time of writing.

Structure:

repo_name/
    manage.py
    base/
    base/__init__.py
    base/app.py
    base/runcelery.py
    base/celeryconfig.py
    base/utility/celery_util.py
    base/tasks/workers.py

So base is the main application package in this example. In the base/__init__.py we create the celery instance as below:

from celery import Celery
celery = Celery('base', config_source='base.celeryconfig')

The base/app.py file contains the flask app factory create_app and note the init_celery(app, celery) it contains:

from base import celery
from base.utility.celery_util import init_celery

def create_app(config_obj):
    """An application factory, as explained here:
    http://flask.pocoo.org/docs/patterns/appfactories/.
    :param config_object: The configuration object to use.
    """
    app = Flask('base')
    app.config.from_object(config_obj)
    init_celery(app, celery=celery)
    register_extensions(app)
    register_blueprints(app)
    register_errorhandlers(app)
    register_app_context_processors(app)
    return app

Moving on to base/runcelery.py contents:

from flask.helpers import get_debug_flag
from base.settings import DevConfig, ProdConfig
from base import celery
from base.app import create_app
from base.utility.celery_util import init_celery
CONFIG = DevConfig if get_debug_flag() else ProdConfig
app = create_app(CONFIG)
init_celery(app, celery)

Next, the base/celeryconfig.py file (as an example):

# -*- coding: utf-8 -*-
"""
Configure Celery. See the configuration guide at ->
http://docs.celeryproject.org/en/master/userguide/configuration.html#configuration
"""

## Broker settings.
broker_url = 'pyamqp://guest:guest@localhost:5672//'
broker_heartbeat=0

# List of modules to import when the Celery worker starts.
imports = ('base.tasks.workers',)

## Using the database to store task state and results.
result_backend = 'rpc'
#result_persistent = False

accept_content = ['json', 'application/text']

result_serializer = 'json'
timezone = "UTC"

# define periodic tasks / cron here
# beat_schedule = {
#    'add-every-10-seconds': {
#        'task': 'workers.add_together',
#        'schedule': 10.0,
#        'args': (16, 16)
#    },
# }

Now define the init_celery in the base/utility/celery_util.py file:

# -*- coding: utf-8 -*-

def init_celery(app, celery):
    """Add flask app context to celery.Task"""
    TaskBase = celery.Task
    class ContextTask(TaskBase):
        abstract = True
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
    celery.Task = ContextTask

For the workers in base/tasks/workers.py:

from base import celery as celery_app
from flask_security.utils import config_value, send_mail
from base.bp.users.models.user_models import User
from base.extensions import mail # this is the flask-mail

@celery_app.task
def send_async_email(msg):
    """Background task to send an email with Flask-mail."""
    #with app.app_context():
    mail.send(msg)

@celery_app.task
def send_welcome_email(email, user_id, confirmation_link):
    """Background task to send a welcome email with flask-security's mail.
    You don't need to use with app.app_context() here. Task has context.
    """
    user = User.query.filter_by(id=user_id).first()
    print(f'sending user {user} a welcome email')
    send_mail(config_value('EMAIL_SUBJECT_REGISTER'),
              email,
              'welcome', user=user,
              confirmation_link=confirmation_link) 

Then, you need to start the celery beat and celery worker in two different cmd prompts from inside the repo_name folder.

In one cmd prompt do a celery -A base.runcelery:celery beat and the other celery -A base.runcelery:celery worker.

Then, run through your task that needed the flask context. Should work.

Solution 2

Flask-mail needs the Flask application context to work correctly. Instantiate the app object on the celery side and use app.app_context like this:

with app.app_context():
    celery.start()

Solution 3

I don't have any points, so I couldn't upvote @codegeek's above answer, so I decided to write my own since my search for an issue like this was helped by this question/answer: I've just had some success trying to tackle a similar problem in a python/flask/celery scenario. Even though your error was from trying to use mail while my error was around trying to use url_for in a celery task, I suspect the two were related to the same problem and that you would have had errors stemming from the use of url_for if you had tried to use that before mail.

With no context of the app present in a celery task (even after including an import app from my_app_module) I was getting errors, too. You'll need to perform the mail operation in the context of the app:

from module_containing_my_app_and_mail import app, mail    # Flask app, Flask mail
from flask.ext.mail import Message    # Message class

@celery.task
def send_forgot_email(email, ref):
    with app.app_context():    # This is the important bit!
        msg = Message("Recover your Crusade Gaming Account")
        msg.recipients = [email]
        msg.sender = "Crusade Gaming [email protected]"
        msg.html = \
        """
        Hello Person,<br/>
        You have requested your password be reset. <a href="{0}" >Click here recover your account</a> or copy and paste this link in to your browser: {0} <br />
        If you did not request that your password be reset, please ignore this.
        """.format(url_for('account.forgot', ref=ref, _external=True))

        mail.send(msg)

If anyone is interested, my solution for the problem of using url_for in celery tasks can be found here

Solution 4

In your mail.py file, import your "app" and "mail" objects. Then, use request context. Do something like this:

from whateverpackagename import app
from whateverpackagename import mail

@celery.task
def send_forgot_email(email, ref):
    with app.test_request_context():
        msg = Message("Recover your Crusade Gaming Account")
        msg.recipients = [email]
        msg.sender = "Crusade Gaming [email protected]"
        msg.html = \
        """
        Hello Person,<br/>
        You have requested your password be reset. <a href="{0}" >Click here recover your account</a> or copy and paste this link in to your browser: {0} <br />
        If you did not request that your password be reset, please ignore this.
        """.format(url_for('account.forgot', ref=ref, _external=True))

        mail.send(msg)

View more solutions

Comments

[Spuds]

mail is the flask_mail instance. mail gets initiated from a different file when the app gets started.

[Jonas Geiregat]

I don't know which solution would be better, adding the context to the whole celery application instance or just the callback function. But you can read all about the Flask Application Context at flask.pocoo.org/docs/appcontext

[Spuds]

How do i give celery access to the flask app? I have them in separate files right now, is that wrong?

[David K. Hess]

Import the app into the celery file like you would do when running Flask. You might have to post your __init__.py for your Flask app or include more details of your setup for me to be more specific.

[Pedro Werneck]

Having the same problem, but instantiating the app and starting the celery inside the context like that isn't working. maybe because the task instance was created outside of the context?

[David K. Hess]

This will only work in the current thread. It's possible you have something in running in another thread. Just do the same thing in that one.

[jaapz]

I think using test_request_context in a non-test environment is not a pretty good idea.

[Dag Høidahl]

This solved a problem when using Flask-Babel in combination with celery. Flask-Babel won't load any translations without a request (because it caches the translations on the request context). Apart from that, Flask-Babel is able to work fine without the request. So using test_request_context() is just a simple way of constructing a functional context, even though it might be a little wasteful.

[gaozhidf]

It has bothered me for a few days. Thanks a lot

[Minh Hoàng]

Why need to run two celery processes? Thank you!

[Bob Jordan]

@MinhHoàng the celery -A base.runcelery: celery beat runs the periodic tasks / cron jobs, only if you have set them up in the base.celeryconfig.py file, specifically refer to beat_schedule variable which I have commented out in the example. If you don't need to use any periodic tasks then you do not need to run two celery processes. You can just run the celery -A base.runcelery: celery worker process.

[Naman]

Does the init_celery(app, celery=celery) inside create_app function do anything?

[Bob Jordan]

@Naman yes it adds the flask application context to celery.Task

[Matheus Torquato]

Won't that create a circular import? You would need to import the snippet above from your app, right?