Trying to send scheduled emails with flask_mail and apscheduler, getting "RuntimeError: Working outside of application context"

Question

I have a Flask app with a rather convoluted structure. The app is constructed with:

server.py

from flask import Flask, jsonify
from flask_cors import CORS
from flask_mail import Mail

from setup.api import api
from setup.db import db
from exceptions import BaseServiceError
from views import auth_ns, users_ns, reminders_ns, whatsapp_bot_ns
from apscheduler.schedulers.background import BackgroundScheduler


def base_service_error_handler(exception: BaseServiceError):
    return jsonify({'error': str(exception)}), exception.code


def create_app(config_obj):
    app = Flask(__name__)
    app.config.from_object(config_obj)

    global mail
    mail = Mail(app)

    global scheduler
    scheduler = BackgroundScheduler(daemon=True)
    scheduler.start()
    
    cors = CORS(app=app)
    cors.init_app(app)
    app.config['CORS_HEADERS'] = 'Content-Type'

    db.init_app(app)
    api.init_app(app)

    api.add_namespace(auth_ns)
    api.add_namespace(users_ns)
    api.add_namespace(reminders_ns)
    api.add_namespace(whatsapp_bot_ns)

    app.register_error_handler(BaseServiceError, base_service_error_handler)

    return app

run.py

from config import config
from models import User, Reminder
from server import create_app, db

app = create_app(config)

@app.shell_context_processor
def shell():
    return {
        "db": db,
        "User": User,
        "Reminder": Reminder
    }

And also other modules like the config. I don't see them being relevant to the problem, so I decided not to include them. I am ready to add any additional information on the project you need.

This app is built to send email notifications at a specific datetime. Here's the module with sending functions:

tools/mailing.py

from datetime import timedelta
from flask_mail import Message
import server

from config import BaseConfig

config = BaseConfig()

def send_first_notification(event_name, user_email):
    message = Message(
        f'Your event {event_name} is beginning in an hour!',
        body='hi',
        sender='[email protected]',
        recipients=[user_email]
        )
    server.mail.send(message=message)


def send_second_notification(event_name, user_email):
    message = Message(
        f'Your event {event_name} is beginning in five minutes!',
        body='hi',
        sender='[email protected]',
        recipients=[user_email]
        )
    server.mail.send(message=message)


def schedule_notifications(datetime, event_name, user_email):

    notification_datetime = datetime

    notification_first = notification_datetime - timedelta(hours=1)
    notification_second = notification_datetime - timedelta(minutes=5)

    server.scheduler.add_job(send_first_notification, run_date=notification_first, args=[event_name, user_email], misfire_grace_time=3600)
    server.scheduler.add_job(send_second_notification, run_date=notification_second, args=[event_name, user_email], misfire_grace_time=3600)

When trying to run it, I recieve the error "RuntimeError: Working outside of application context" on the server.mail.send(message=message) command.

How can I fix this?

I've read I should add with app.app_context() but I can't import the app from the app constructing modules. Adding app = current_app._get_current_object() raises the same error: "RuntimeError: Working outside of application context." I've also tried 'current_app = LocalProxy(_find_app)', it results in "NameError: name '_find_app' is not defined".

Answer 1

Not sure if the question is still current, however I had a similar problem. For me it works the following way:

I initialize both the scheduler and the mail in extensions.py:

""" This module contains extensions """
from flask_apscheduler import APScheduler as _BaseAPScheduler
from flask_mail import Mail

# Subclass APScheduler to provide the flask app context for all jobs
class APScheduler(_BaseAPScheduler):
    def run_job(self, tmp_id, jobstore=None):
        with self.app.app_context():
            super().run_job(id=tmp_id, jobstore=jobstore)


# Instantiate scheduler and mail object
scheduler = APScheduler()
mail = Mail()

In my app factory (__init__.py), I load the extensions (and some variables from the config class) to build the app like this (note the passed app_context() when building the jobs for the scheduler):

""" Init file for the Flask app. """
import atexit

import os

from flask import Flask

from my_app.config import Config
from my_app.extensions import scheduler, mail
from my_app.main.scheduler_utils import add_tasks


def create_app(config_class=Config):
    """
    Function for Flask factory pattern
    :param config_class: Class holding all the flask config variables
    :return: Configured app instance
    """

    # Flask instance
    app = Flask(__name__, static_url_path='/my_app/static')
    app.config.from_object(config_class)

    # Mail lib
    mail.init_app(app)

    # Instantiate scheduler, establish app context
    scheduler.init_app(app)

    # Add jobs
    with app.app_context():
        add_tasks(some kwargs)

    # Add closing events for scheduler
    scheduler.start()
    if scheduler.running:
        atexit.register(scheduler.shutdown)

    return app

Finally, inside on of my jobs I send emails using a send_mail() function, that looks like this in a simplified version, stored in utils.py:

""" File with utility functions """
import datetime
import logging
import os

import jinja2
from flask_mail import Message

from my_app.config import Config
from my_app.extensions import mail, scheduler

# Global logger instances
flask_logger = logging.getLogger('Flask_logger')
mail_logger = logging.getLogger('mail_logger')

def send_mail(mail_template: str, email_params: dict, send_to: list, subject: str,
              sent_from: str = Config.DEFAULT_SENDER) -> None:
    """
    Function to send the mail
    @param mail_template: str, the html file name of the template
    @param email_params: dict, kwargs to be rendered by Jinja within the mail_template file
    @param sent_from: str, the sender to be shown
    @param send_to: list, the list of all recipients
    @param subject: str, subject
    @return: None
    """
    try:
        if not send_to:
            # If the recipient list is empty, do not send the email
            mail_logger.warning('Recipient list is empty. Email with subject %s not sent.', subject)
            return

        with mail.app.app_context():
            mail_logo = os.path.join(Config.LOGO_FOLDER, 'your_cool_mail_logo.png')
            msg = Message(subject=subject, sender=sent_from, recipients=send_to)
    
            # Attempt to load the template, handling TemplateNotFound
            try:
                template = jinja2.Environment(
                    loader=jinja2.FileSystemLoader(Config.MAIL_TEMPLATES)).get_template(mail_template)
            except jinja2.exceptions.TemplateNotFound:
                flask_logger.error('Template not found: %s', mail_template)
                return
    
            msg.html = template.render(**email_params)
    
            with open(mail_logo, 'rb') as image_file:
                image_data = image_file.read()
                msg.attach('a_cool_name_for_your_logo.png', 'image/png', image_data,
                           'inline', headers=[['Content-ID', '<corporate_logo>']])
                
            mail.send(msg)
        mail_logger.debug('Mail with subject "%s" sent successfully.', subject)

    except Exception as err:
        mail_logger.error('An error occurred at function "send_mail": %s', err, exc_info=True)

Note again the with scheduler.app.app_context() part which provides the missing app context inside the job routine. For sending nice messages, I use html templates where I pass variables and render the template using jinja, of course this can be simplified and also the logo part is optional.

Let me know if it helps please.

Answer 2

A similar approach worked for me as well. I created a separate module where I initiated the instances of APScheduler and Mail:

# host_services.py

from flask_mail import Mail
from flask_apscheduler import APScheduler

scheduler = APScheduler()
mail = Mail()

Then I imported the instances in the file with the app factory and called for their init methods:

# server.py

from flask import Flask
from flask_restx import Api
from project.host_services import mail, scheduler
from project.views.reminder import reminder_ns

def create_app(config):
    app = Flask(__name__)
    app.config.from_object(config)
    scheduler.init_app(app)
    scheduler.start()
    mail.init_app(app)
    
    return app

And then I was able to use them like this in my mailing service (only posting the corresponding methods):

# mailing.py

from flask import current_app
from project.host_services import mail, scheduler

app = current_app._get_current_object()

def _send_notification(self, title_text: str, body_text: str) -> None:
    """
    Sends an email.
    """
    message = Message(
        title_text,
        body=body_text,
        sender='[email protected]',
        recipients=[self.reminder.user_email]
        )
    with app.app_context():
        mail.send(message=message)

def _create_scheduler_job(self, run_date: datetime, text: str) -> None:
    """
    Schedules a job to send a notification email.
    """
    scheduler.add_job(id=None, func=self._send_notification,
                        run_date=run_date,
                        args=[self._make_email_title(text),
                        self._make_email_body()],
                        misfire_grace_time=3600)

But I had to get the app instance from flask by using the current_app's _get_current_object() method to finally be able to make the Mail instance work in the app context. Calling it from the Mail instance didn't work for me.

I hope this is comprehensible.