Flask-mail send_async_email() generates an exception and RunTimeError: Working outside of application context
Flask-mail send_async_email() generates an exception and RunTimeError: Working outside of application context
除 send_email() 外,其他一切正常。无论我使用 localhost
还是远程主机,我都会遇到同样的异常。我的猜测是我没有将上下文推送到正确的位置,但我已经盯着它看了很长时间,但我显然没有看到错误。任何帮助将不胜感激!
__init__.py:
from flask import Flask, Blueprint, jsonify, ...
from config import config
from flask_login import LoginManager
from extensions import db, mail, moment
login_manager = LoginManager()
login_manager.login_view = 'auth.login'
def create_app():
app = Flask(__name__,
static_url_path='',
static_folder='../app/static',
template_folder='../app/templates')
app.config.from_object(config['default'])
config['default'].init_app(app)
db.init_app(app)
mail.init_app(app)
moment.init_app(app)
with app.app_context():
db.create_all()
migrate = Migrate(app, db)
from app import models
from .templates.auth import auth_bp
from .templates.main import main_bp
app.register_blueprint(main_bp)
app.register_blueprint(auth_bp, url_prefix='/auth')
login_manager.init_app(app)
return app
views.py:
from flask import abort, Blueprint, flash, ...
from flask.globals import current_app
from flask_login import current_user, login_required, login_user, logout_user
from werkzeug.security import generate_password_hash
from datetime import datetime, timezone
from wtforms.validators import UUID
from . import auth_bp
from extensions import db, common_context
from app.models. ...
from .forms import LoginForm, RegistrationForm, ...
from .email import send_email
@auth_bp.route('/register/', methods=['GET', 'POST'], defaults={'user_id': None})
def register(user_id):
from sqlalchemy.exc import IntegrityError
from .email import send_email
if current_user.is_authenticated:
return redirect(url_for('main.home'))
form=RegistrationForm()
if form.validate_on_submit():
try:
individual = Individual(...
)
db.session.add(individual)
db.session.flush()
individual_email = Individual_email(...
)
db.session.add(individual_email)
user = User(...
)
db.session.add(user)
db.session.commit()
token = user.generate_confirmation_token()
# with current_app.app_context():
# print(individual_email.email, user, individual.first_name, token) - this works!!!
send_email(individual_email.email, 'Please Confirm Your Account',
'auth/email/confirm', user=user, token=token, individual=individual)
flash('A confirmation email has been sent to you by email.')
except AssertionError as err:
db.session.rollback()
abort(409, err)
except IntegrityError as err:
db.session.rollback()
abort(409, err.orig)
except Exception as err:
db.session.rollback()
abort(500, err)
finally:
db.session.close()
return redirect(url_for('auth.login'))
context = {
'form': form,
'title': 'Registration',
}
return render_template('auth/register.html', **context, **common_context)
email.py:
from flask import render_template, current_app
from flask_mail import Message
from threading import Thread
from app import mail
def send_async_email(app, msg):
with app.app_context():
mail.send(msg)
def send_email(to, subject, template, **kwargs):
app=current_app
msg = Message(app.config['MAIL_SUBJECT_PREFIX'] + subject, sender = app.config['MAIL_SENDER'], recipients=[to])
msg.body = render_template(template + '.txt', **kwargs)
msg.html = render_template(template + '.html', **kwargs)
thr = Thread(target=send_async_email, args=[app, msg])
thr.start()
return thr
配置:
import os
from dotenv import load_dotenv
basedir = os.path.abspath(os.path.dirname(__file__))
load_dotenv(os.path.join(basedir, '.env'))
class Config(object):
DEBUG = False
TESTING = False
CSRF_ENABLED = True
SECRET_KEY = os.environ.get('SECRET_KEY')
SQLALCHEMY_TRACK_MODIFICATIONS = False
...
@staticmethod
def init_app(app):
pass
class ProductionConfig(Config):
DEBUG = False
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URI')
MAIL_SERVER = os.environ.get('MAIL_SERVER', 'smtp.gmail.com')
MAIL_PORT = int(os.environ.get('MAIL_PORT', '587'))
MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() in \
['true', 'on', '1']
MAIL_USE_SSL = os.environ.get('MAIL_USE_SSL', 'false').lower()
MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
MAIL_SUBJECT_PREFIX = '[...]'
MAIL_SENDER = '... <...>'
MAIL_ADMIN = os.environ.get('MAIL_ADMIN')
class DevelopmentConfig(Config):
DEVELOPMENT = True
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URI')
MAIL_SERVER = 'localhost'
MAIL_PORT = 25
MAIL_USE_TLS = False
MAIL_USE_SSL = False
# MAIL_DEBUG = app.debug
MAIL_USERNAME = '...'
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
MAIL_DEFAULT_SENDER = '... <...>'
MAIL_MAX_EMAILS = None
# MAIL_SUPPRESS_SEND = app.testing
MAIL_ASCII_ATTACHMENTS = False
MAIL_SUBJECT_PREFIX = '[...]'
MAIL_SENDER = '... <...>'
MAIL_ADMIN = '...'
config = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig,
'default': DevelopmentConfig
}
输出:
Exception in thread Thread-23:
Traceback (most recent call last):
File "C:\...\Python38\lib\threading.py", line 932, in _bootstrap_inner
127.0.0.1 - - [11/Feb/2021 21:04:21] "POST /auth/register/ HTTP/1.1" 302 -
self.run()
File "C:\...\Python38\lib\threading.py", line 870, in run
self._target(*self._args, **self._kwargs)
File "C:\...\app\templates\auth\email.py", line 7, in send_async_email
with app.app_context():
127.0.0.1 - - [11/Feb/2021 21:04:21] "GET /auth/login HTTP/1.1" 200 -
File "c:\...\venv\lib\site-packages\werkzeug\local.py", line 347, in __getattr__
return getattr(self._get_current_object(), name)
File "c:\...\venv\lib\site-packages\werkzeug\local.py", line 306, in _get_current_object
return self.__local()
File "c:\...\venv\lib\site-packages\flask\globals.py", line 52, in _find_app
raise RuntimeError(_app_ctx_err_msg)
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 some way. To solve
this, set up an application context with app.app_context(). See the
documentation for more information.
将 email.py
中的两个函数合并为一个以避免上下文问题:
def send_email(to, subject, template, **kwargs):
app=current_app
with app.app_context():
msg = Message(app.config['MAIL_SUBJECT_PREFIX'] + subject, sender =
app.config['MAIL_SENDER'], recipients=[to])
msg.body = render_template(template + '.txt', **kwargs)
msg.html = render_template(template + '.html', **kwargs)
Thread(target=mail.send(msg)).start()
第一次尝试时效果很好。我查看了其他 Whosebug 帖子并收到了将 create_app
导入此模块的建议。我反对这条路线。我有两个实例而不是一个或相同的异常,具体取决于 create_app()
调用的位置。
除 send_email() 外,其他一切正常。无论我使用 localhost
还是远程主机,我都会遇到同样的异常。我的猜测是我没有将上下文推送到正确的位置,但我已经盯着它看了很长时间,但我显然没有看到错误。任何帮助将不胜感激!
__init__.py:
from flask import Flask, Blueprint, jsonify, ...
from config import config
from flask_login import LoginManager
from extensions import db, mail, moment
login_manager = LoginManager()
login_manager.login_view = 'auth.login'
def create_app():
app = Flask(__name__,
static_url_path='',
static_folder='../app/static',
template_folder='../app/templates')
app.config.from_object(config['default'])
config['default'].init_app(app)
db.init_app(app)
mail.init_app(app)
moment.init_app(app)
with app.app_context():
db.create_all()
migrate = Migrate(app, db)
from app import models
from .templates.auth import auth_bp
from .templates.main import main_bp
app.register_blueprint(main_bp)
app.register_blueprint(auth_bp, url_prefix='/auth')
login_manager.init_app(app)
return app
views.py:
from flask import abort, Blueprint, flash, ...
from flask.globals import current_app
from flask_login import current_user, login_required, login_user, logout_user
from werkzeug.security import generate_password_hash
from datetime import datetime, timezone
from wtforms.validators import UUID
from . import auth_bp
from extensions import db, common_context
from app.models. ...
from .forms import LoginForm, RegistrationForm, ...
from .email import send_email
@auth_bp.route('/register/', methods=['GET', 'POST'], defaults={'user_id': None})
def register(user_id):
from sqlalchemy.exc import IntegrityError
from .email import send_email
if current_user.is_authenticated:
return redirect(url_for('main.home'))
form=RegistrationForm()
if form.validate_on_submit():
try:
individual = Individual(...
)
db.session.add(individual)
db.session.flush()
individual_email = Individual_email(...
)
db.session.add(individual_email)
user = User(...
)
db.session.add(user)
db.session.commit()
token = user.generate_confirmation_token()
# with current_app.app_context():
# print(individual_email.email, user, individual.first_name, token) - this works!!!
send_email(individual_email.email, 'Please Confirm Your Account',
'auth/email/confirm', user=user, token=token, individual=individual)
flash('A confirmation email has been sent to you by email.')
except AssertionError as err:
db.session.rollback()
abort(409, err)
except IntegrityError as err:
db.session.rollback()
abort(409, err.orig)
except Exception as err:
db.session.rollback()
abort(500, err)
finally:
db.session.close()
return redirect(url_for('auth.login'))
context = {
'form': form,
'title': 'Registration',
}
return render_template('auth/register.html', **context, **common_context)
email.py:
from flask import render_template, current_app
from flask_mail import Message
from threading import Thread
from app import mail
def send_async_email(app, msg):
with app.app_context():
mail.send(msg)
def send_email(to, subject, template, **kwargs):
app=current_app
msg = Message(app.config['MAIL_SUBJECT_PREFIX'] + subject, sender = app.config['MAIL_SENDER'], recipients=[to])
msg.body = render_template(template + '.txt', **kwargs)
msg.html = render_template(template + '.html', **kwargs)
thr = Thread(target=send_async_email, args=[app, msg])
thr.start()
return thr
配置:
import os
from dotenv import load_dotenv
basedir = os.path.abspath(os.path.dirname(__file__))
load_dotenv(os.path.join(basedir, '.env'))
class Config(object):
DEBUG = False
TESTING = False
CSRF_ENABLED = True
SECRET_KEY = os.environ.get('SECRET_KEY')
SQLALCHEMY_TRACK_MODIFICATIONS = False
...
@staticmethod
def init_app(app):
pass
class ProductionConfig(Config):
DEBUG = False
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URI')
MAIL_SERVER = os.environ.get('MAIL_SERVER', 'smtp.gmail.com')
MAIL_PORT = int(os.environ.get('MAIL_PORT', '587'))
MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() in \
['true', 'on', '1']
MAIL_USE_SSL = os.environ.get('MAIL_USE_SSL', 'false').lower()
MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
MAIL_SUBJECT_PREFIX = '[...]'
MAIL_SENDER = '... <...>'
MAIL_ADMIN = os.environ.get('MAIL_ADMIN')
class DevelopmentConfig(Config):
DEVELOPMENT = True
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URI')
MAIL_SERVER = 'localhost'
MAIL_PORT = 25
MAIL_USE_TLS = False
MAIL_USE_SSL = False
# MAIL_DEBUG = app.debug
MAIL_USERNAME = '...'
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
MAIL_DEFAULT_SENDER = '... <...>'
MAIL_MAX_EMAILS = None
# MAIL_SUPPRESS_SEND = app.testing
MAIL_ASCII_ATTACHMENTS = False
MAIL_SUBJECT_PREFIX = '[...]'
MAIL_SENDER = '... <...>'
MAIL_ADMIN = '...'
config = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig,
'default': DevelopmentConfig
}
输出:
Exception in thread Thread-23:
Traceback (most recent call last):
File "C:\...\Python38\lib\threading.py", line 932, in _bootstrap_inner
127.0.0.1 - - [11/Feb/2021 21:04:21] "POST /auth/register/ HTTP/1.1" 302 -
self.run()
File "C:\...\Python38\lib\threading.py", line 870, in run
self._target(*self._args, **self._kwargs)
File "C:\...\app\templates\auth\email.py", line 7, in send_async_email
with app.app_context():
127.0.0.1 - - [11/Feb/2021 21:04:21] "GET /auth/login HTTP/1.1" 200 -
File "c:\...\venv\lib\site-packages\werkzeug\local.py", line 347, in __getattr__
return getattr(self._get_current_object(), name)
File "c:\...\venv\lib\site-packages\werkzeug\local.py", line 306, in _get_current_object
return self.__local()
File "c:\...\venv\lib\site-packages\flask\globals.py", line 52, in _find_app
raise RuntimeError(_app_ctx_err_msg)
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 some way. To solve
this, set up an application context with app.app_context(). See the
documentation for more information.
将 email.py
中的两个函数合并为一个以避免上下文问题:
def send_email(to, subject, template, **kwargs):
app=current_app
with app.app_context():
msg = Message(app.config['MAIL_SUBJECT_PREFIX'] + subject, sender =
app.config['MAIL_SENDER'], recipients=[to])
msg.body = render_template(template + '.txt', **kwargs)
msg.html = render_template(template + '.html', **kwargs)
Thread(target=mail.send(msg)).start()
第一次尝试时效果很好。我查看了其他 Whosebug 帖子并收到了将 create_app
导入此模块的建议。我反对这条路线。我有两个实例而不是一个或相同的异常,具体取决于 create_app()
调用的位置。