Flask：如何在完整路由代码执行之前返回验证

Question

抱歉新手问题 - 我是 Python 新手。

在主代码执行之前我必须返回 Flask 路由参数的验证结果：

@app.route('/', methods=['POST'])
def index():
    validation = code_to_check_parameters()
    if validation:
        return make_response('Valid', 200)
    else:
        return make_response('Invalid', some_error_code)

    # this line is unreachable because of returns, but required
    video = request.files['video']
    result = heavy_video_processing(video) # takes 4-5 minutes, so I can't wait for this
    r = requests.post(URL, results)
    return r.status_code

期望什么：

• 检查路由参数并返回验证结果
• 通过requests.post重度执行4-5分钟后继续执行Python代码并推送视频处理结果。

如何做？

Answer 1

您正在寻找的称为“异步”处理。换句话说，第一个请求很早就存在，但该过程在后台继续进行，并且需要将来的请求来检查结果。

首先，请注意第一个请求应返回状态 202 而不是 200：

202（已接受）状态码表示请求已被接受处理，但处理尚未完成。 (RFC 9110)

另请注意，您可以在做出响应之前生成一个新的 python 线程。该线程可以执行长进程，并将结果存储在可由flask访问的共享变量（或数据库）中。

使用第一个响应 (202) 的“location”标头来指示稍后应使用哪个 url 来获取结果也很好。

完整代码如下：

import threading
import time
import random
import uuid

from flask import Flask, request, make_response


class VideoProcessor:
    def __init__(self):
        self.results = {}

    def _procees_video(self, video, id):
        time.sleep(10)  # fake long process
        result = random.randint(0, 1000)  # fake result
        self.results[id] = result

    def start_processing(self, video):
        id = uuid.uuid4().hex[:8]  # generate random unique id so the user can track the status
        thread = threading.Thread(target=self._procees_video, args=(video, id))
        thread.start()
        self.results[id] = "PROCESSING"
        return id

video_processor = VideoProcessor()
app = Flask(__name__)


def code_to_check_parameters():
    # some code to check parameters
    return True


@app.route('/', methods=['POST'])
def index():
    validation = code_to_check_parameters()
    if not validation:
        return make_response('Invalid', 400)
    video = request.files['video']
    id = video_processor.start_processing(video)
    return make_response(
        f'Valid, id = {id}',
        202,
        {'Location': f'/status/{str(id)}'}
    )


@app.route('/status/<id>', methods=['GET'])
def status(id):
    print(video_processor.results)
    print(id)
    if id not in video_processor.results:
        return make_response('Not found', 404)
    if video_processor.results[id] == "PROCESSING":
        return make_response('Still processing', 202)
    return make_response(str(video_processor.results[id]), 200)


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

请注意，此代码使用了 python 提供的最简单的工具。需要添加线程安全、崩溃恢复、资源限制等措施才能获得稳健的结果。