redis 发送确认电子邮件与 flask ,RQ(工作以外的应用程序上下文.)

bz4sfanl  于 2023-01-01  发布在  Redis
关注(0)|答案(1)|浏览(160)

我尝试在新用户注册时发送确认电子邮件。当我发送确认电子邮件时,一切正常,但是,当我尝试在后台运行该任务时,我收到以下错误:

File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/rq/worker.py", line 1075, in perform_job
    rv = job.perform()
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/rq/job.py", line 854, in perform
    self._result = self._execute()
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/rq/job.py", line 877, in _execute
    result = self.func(*self.args, **self.kwargs)
  File "/Users/martifont/Dev/bz/./FlaskApp/utils.py", line 22, in send_confirmation_email
    msg = Message()
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/flask_mail.py", line 273, in __init__
    sender = sender or current_app.extensions['mail'].default_sender
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/werkzeug/local.py", line 316, in __get__
    obj = instance._get_current_object()  # type: ignore[misc]
  File "/Users/martifont/Dev/bz/venv/lib/python3.10/site-packages/werkzeug/local.py", line 509, in _get_current_object
    raise RuntimeError(unbound_message) from None
RuntimeError: Working outside of application context.

This typically means that you attempted to use functionality that needed
the current application. To solve this, set up an application context
with app.app_context(). See the documentation for more information.

我试过在后台运行一个不同的任务,完全没有问题。

初始化. py

import os
import redis
from rq import Queue
from flask import Flask, Blueprint, request, redirect, url_for
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, current_user
from flask_mail import Mail
from flask_admin import Admin, AdminIndexView
from flask_admin.contrib.sqla import ModelView
from flask_migrate import Migrate

db = SQLAlchemy()
migrate = Migrate()
mail = Mail()
login_manager = LoginManager()
admin = Admin(template_mode='bootstrap3')

r = redis.Redis()
q = Queue(connection=r)

def create_app():
app = Flask(__name__)

    #code here...
    
    with app.app_context():
        #IMPORT BLUEPRINTS    
        from .main import main as main_blueprint
        app.register_blueprint(main_blueprint)
    
        from .users import users as users_blueprint
        app.register_blueprint(users_blueprint)
    
    return app

users.py

import os
import secrets
from PIL import Image
from flask import Blueprint, render_template, redirect, url_for, request, flash

from flask import current_app as app

from werkzeug.security import generate_password_hash, check_password_hash
from flask_login import login_user, logout_user, login_required, current_user
from FlaskApp import db, mail, r, q
from flask_mail import Message
from .models import User
from .forms import SignupForm, LoginForm, RequestResetForm, ResetPasswordForm, UpdateAccountForm
from .utils import send_reset_email, send_confirmation_email, example

users = Blueprint('users', __name__)

#SIGNUP
@users.route('/signup', methods=\['GET', 'POST'\])
def signup():
form = SignupForm()
if form.validate_on_submit():
user = User(email=form.email.data, username=form.username.data, confirmed=False, is_admin=False, password = generate_password_hash(form.password.data, method='sha256'))
db.session.add(user)
db.session.commit()

        user_id=user.id
        with app.app_context():
            job = q.enqueue(send_confirmation_email, user)
    
        login_user(user)
        flash('A confirmation email has been sent to your inbox.', 'is-success')
        return redirect(url_for('users.account', id=user.id))
    return render_template('signup.html', form=form)

utils.py

import os
import secrets
from flask import url_for
from flask_mail import Message
from FlaskApp import mail

def send_confirmation_email(user):
    token = user.get_confirmation_token()
    msg = Message()
    msg.subject = ('Email Confirmation')
    msg.sender = 'MAIL_DEFAULT_SENDER'
    msg.recipients = [user.email]
    msg.body = f'''
Welcome { user.username }!,
Thanks for signing up. Please follow this link to activate your account:
{url_for('users.confirm_email', token=token, _external=True)}
Thanks!

'''
    mail.send(msg)
pgky5nke

pgky5nke1#

当您在视图或creat_app工厂函数中时,不需要使用app_context()上下文管理器。
回溯定义了错误发生的位置,也就是您示例化msg=Message()变量的时间,因此,我建议您不要调用视图中的应用上下文,而是重构视图和发送函数,如下所示:
users.py

from flask import current_app

@users.route('/signup', methods=\['GET', 'POST'\])
def signup():
    form = SignupForm()
    if form.validate_on_submit():
        user = User(email=form.email.data, username=form.username.data, confirmed=False, is_admin=False, password = generate_password_hash(form.password.data, method='sha256'))
        db.session.add(user)
        db.session.commit()

        user_id=user.id
        # pass the current app object to the send_confirmation_email function.
        # Its important to pass the object and no the proxy
        app = current_app._get_current_object() 
        job = q.enqueue(send_confirmation_email, args=(app, user))
    
        login_user(user)
        flash('A confirmation email has been sent to your inbox.', 'is-success')
        return redirect(url_for('users.account', id=user.id))
    return render_template('signup.html', form=form)

utils.py

def send_confirmation_email(app, user):

    with app.app_context():
        token = user.get_confirmation_token()
        msg = Message()
        msg.subject = ('Email Confirmation')
        msg.sender = 'MAIL_DEFAULT_SENDER'
        msg.recipients = [user.email]
        msg.body = f'''
            Welcome { user.username }!,
            Thanks for signing up. Please follow this link to activate your account:
            {url_for('users.confirm_email', token=token, _external=True)}
            Thanks!'''
        mail.send(msg)

相关问题