Flask-Mail - 基於Flask-Cookiecutter異步發送電子郵件

Question

我的燒瓶項目基於Flask-Cookiecutter，我需要異步發送電子郵件。

功能發送電子郵件配置Miguel’s Tutorial和同步發送工作正常，但我不知道，我怎麼能修改它發送異步。

我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 

我view.py在我的本地主機，它的起始命令view.py

@blueprint.route('/mailer', methods=['GET', 'POST']) 
def mailer(): 
    user = current_user.full_name 
    send_email(('name@gmail.com'), 
       'New mail', 'test.html', 
       user=user) 
    return "Mail has been send." 

應用程序運行

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) 

路線：

python manage.py server 

當我打電話功能用於發送郵件，在控制檯輸出爲：

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. 

感謝您的任何答覆。

Answer 1

好吧，我發現我的問題解決方案，我在這裏張貼他人開發商：

我創建文件：email.py的代碼：

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 

我view.py：

... 
from app.email import send_email 
... 

@blueprint.route('/mailer', methods=['GET', 'POST']) 
def mailer(): 
    user = current_user.full_name 
    send_email(('name@gmail.com'), 
       'New mail', 'test.html', 
       user=user) 
    return "Mail has been send." 

而當我打電話http://localhost:5000/mailer它開始倒計時，幾秒鐘後發送郵件。

Answer 2

移動電子郵件發送功能，後臺線程：

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 

Answer 3

您可以將app = Flask(__name__)移出應用程序工廠，並將其置於模塊級別。這使您可以將應用程序實例及其應用程序上下文傳遞到您的線程以發送電子郵件。您可能需要在其他區域更改一些導入以防止循環依賴，但它不應該太糟糕。

這是使用Flask-Mail和Flask-RESTful的example of how you can do this。它還展示瞭如何使用pytest進行測試。

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) 

而且你的模塊用於發送電子郵件在你有這樣的事情：

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)