Flask返回响应后执行一个函数

Question

我有一些代码需要在 Flask返回响应后执行.我认为设置像Celery这样的任务队列并不复杂.关键要求是Flask必须在运行此函数之前将响应返回给客户端.它不能等待函数执行.

关于这一点存在一些现有的问题,但是没有一个答案似乎解决了在将响应发送到客户端之后运行任务,它们仍然同步执行然后返回响应.

• 在Flask中返回响应后需要执行一个函数
• Python Flask立即发送响应
• 在Flask中返回响应后需要执行一个函数
• Flask结束响应并继续处理

Answer 1

长话短说,Flask没有提供任何特殊功能来实现这一目标.对于简单的一次性任务,请考虑Python的多线程,如下所示.对于更复杂的配置,请使用RQ或Celery等任务队列.

为什么？

了解Flask提供的功能以及为什么它们无法实现预期目标非常重要.所有这些在其他情况下都很有用,并且阅读良好,但对后台任务没有帮助.

Flask的after_request处理程序

Flask的after_request处理程序,如此延迟请求回调模式中的详细信息以及每个请求附加不同函数的此代码段,将把请求传递给回调函数.预期用例是修改请求,例如附加cookie.

因此,请求将等待这些处理程序完成执行,因为期望是请求本身将因此而改变.

Flask的teardown_request处理程序

这类似于after_request,但teardown_request没有收到request对象.这意味着它不会等待请求,对吧？

这似乎是解决方案,因为类似Stack Overflow问题的答案表明.而且由于Flask的文档解释了拆解回调与实际请求无关并且没有收到请求上下文,因此您有充分的理由相信这一点.

不幸的是,teardown_request它仍然是同步的,它只是在请求不再可修改时在Flask的请求处理的后期部分发生.在返回响应之前,Flask 仍会等待拆卸功能完成,因为这个Flask回调和错误列表指示.

Flask的流式响应

Flask可以通过传递生成器来传递响应Response(),因为此Stack Overflow对类似问题的回答表明.

通过流式传输,客户端确实在请求结束之前开始接收响应.但是,请求仍然会同步运行,因此处理请求的工作人员将忙于流完成.

此流式传输的Flask模式包含一些使用文档stream_with_context(),这是包含请求上下文所必需的.

那么解决方案是什么？

Flask没有提供在后台运行功能的解决方案,因为这不是Flask的责任.

在大多数情况下,解决此问题的最佳方法是使用RQ或Celery等任务队列.这些是管理棘手的事情,比如配置,安排和为你分配工人.这是这类问题最常见的答案,因为它是最正确的,并迫使你以考虑上下文等的方式进行设置.正确.

如果需要在后台运行函数并且不想设置队列来管理它,可以使用内置的Python threading或multiprocessing生成后台工作程序.

您无法request从后台任务访问Flask的线程本地人或其他人,因为请求将不会在那里处于活动状态.而是在创建时将所需的数据从视图传递到后台线程.

@app.route('/start_task')
def start_task():
    def do_work(value):
        # do something that takes a long time
        import time
        time.sleep(value)

    thread = Thread(target=do_work, kwargs={'value': request.args.get('value', 20))
    thread.start()
    return 'started'

Answer 2

Flask是一个WSGI应用程序,因此它在响应后根本无法处理任何事情.这就是为什么不存在这样的处理程序的原因,WSGI应用程序本身只负责构造WSGI服务器的响应迭代器对象.

一个WSGI服务器然而(如gunicorn)可以很容易提供这种功能,但捆绑应用程序服务器是有很多原因一个非常糟糕的主意.

出于这个原因,WSGI提供了中间件规范,Werkzeug提供了许多帮助器来简化常见的中间件功能.其中有一个ClosingIterator类,它允许您将方法挂钩到close响应迭代器的方法,该方法在请求关闭后执行.

这是一个after_response作为Flask扩展完成的天真实现的示例:

import traceback
from werkzeug.wsgi import ClosingIterator

class AfterResponse:
    def __init__(self, app=None):
        self.callbacks = []
        if app:
            self.init_app(app)

    def __call__(self, callback):
        self.callbacks.append(callback)
        return callback

    def init_app(self, app):
        # install extension
        app.after_response = self

        # install middleware
        app.wsgi_app = AfterResponseMiddleware(app.wsgi_app, self)

    def flush(self):
        for fn in self.callbacks:
            try:
                fn()
            except Exception:
                traceback.print_exc()

class AfterResponseMiddleware:
    def __init__(self, application, after_response_ext):
        self.application = application
        self.after_response_ext = after_response_ext

    def __call__(self, environ, after_response):
        iterator = self.application(environ, after_response)
        try:
            return ClosingIterator(iterator, [self.after_response_ext.flush])
        except Exception:
            traceback.print_exc()
            return iterator

您可以像这样使用此扩展程序:

import flask
app = flask.Flask("after_response")
AfterResponse(app)

@app.after_response
def say_hi():
    print("hi")

@app.route("/")
def home():
    return "Success!\n"

当您卷曲"/"时,您会在日志中看到以下内容:

127.0.0.1 - - [24/Jun/2018 19:30:48] "GET / HTTP/1.1" 200 -
hi

这解决了这个问题,只是没有引入任何线程(GIL ??)或必须安装和管理任务队列和客户端软件.

Answer 3

有 3 种方法可以做到这一点，全部有效：

1. 线程

@app.route('/inner')
def foo():
    for i in range(10):
        sleep(1)
        print(i)
    return 

@app.route('/inner', methods=['POST'])
def run_jobs():
    try:
        thread = Thread(target=foo)
        thread.start()
        return render_template("index_inner.html", img_path=DIR_OF_PHOTOS, video_path=UPLOAD_VIDEOS_FOLDER)
                            

2.AfterResponse装饰器

app = Flask(__name__)
AfterResponse(app)

@app.route('/inner', methods=['POST'])
def save_data():
    pass

@app.after_response
def foo():
    for i in range(10):
        sleep(1)
        print(i)
    return 

3. 关闭时调用

from time import sleep

from flask import Flask, Response, request


app = Flask('hello')


@app.route('/')
def hello():

    response = Response('hello')

    @response.call_on_close
    def on_close():
        for i in range(10):
            sleep(1)
            print(i)

    return response


if __name__ == '__main__':
    app.run()

Answer 4

Flask 现在支持（通过 Werkzeug）call_on_close响应对象上的回调装饰器。以下是您如何使用它：

@app.after_request
def response_processor(response):
    # Prepare all the local variables you need since the request context
    # will be gone in the callback function

    @response.call_on_close
    def process_after_request():
        # Do whatever is necessary here
        pass

    return response

好处：

1. call_on_close使用该close方法的 WSGI 规范设置返回响应后调用的函数。

2. 没有线程，没有后台作业，没有复杂的设置。它在同一个线程中运行，而不会阻止请求返回。

缺点：

1. 没有请求上下文或应用上下文。您必须保存所需的变量，以传递到闭包中。
2. 没有本地堆栈，因为所有这些都被拆除了。如果需要，您必须创建自己的应用程序上下文。
3. 如果您尝试写入数据库，Flask-SQLAlchemy 将无声地失败（我还没有弄清楚原因，但可能是由于上下文关闭）。（它有效，但如果您有一个现有对象，则必须使用session.add或将其添加到新会话中session.merge；这不是缺点！）

Answer 5

Flask 蓝图的中间件解决方案

这与 Matthew Story 提出的解决方案相同（恕我直言，这是完美的解决方案 - 感谢 Matthew），适用于 Flask 蓝图。这里的秘诀是使用 current_app 代理获取应用程序上下文。在这里阅读更多信息（http://flask.pocoo.org/docs/1.0/appcontext/）

让我们假设 AfterThisResponse 和 AfterThisResponseMiddleware 类放置在 .utils.after_this_response.py 的模块中

然后在 Flask 对象创建发生的地方，你可能有，例如......

__init__.py

from api.routes import my_blueprint
from .utils.after_this_response import AfterThisResponse

app = Flask( __name__ )
AfterThisResponse( app )
app.register_blueprint( my_blueprint.mod )

然后在您的蓝图模块中...

a_blueprint.py

from flask import Blueprint, current_app

mod = Blueprint( 'a_blueprint', __name__, url_prefix=URL_PREFIX )

@mod.route( "/some_resource", methods=['GET', 'POST'] )
def some_resource():
    # do some stuff here if you want

    @current_app.after_this_response
    def post_process():
        # this will occur after you finish processing the route & return (below):
        time.sleep(2)
        print("after_response")

    # do more stuff here if you like & then return like so:
    return "Success!\n"

Answer 6

除了其他解决方案之外，您还可以通过组合 after_this_request 和 response.call_on_close 来执行路由特定操作：

@app.route('/')
def index():
    # Do your pre-response work here
    msg = 'Hello World!'
    @flask.after_this_request
    def add_close_action(response):
        @response.call_on_close
        def process_after_request():
            # Do your post-response work here
            time.sleep(3.0)
            print('Delayed: ' + msg)
        return response
    return msg