Flask: how to return validation before full route code execution

Question

Sorry for newbie question - I am new in Python.

I have to return validation result of Flask route parameter before main code execution:

@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

What expected:

• Check route parameters and return validation results
• Continue execution of Python code and push video processing results after 4-5 minutes of heavy execution by requests.post.

How to do it?

Answer 1

What you're looking for is called "async" processing. In other words, the first request exists early but the process continues in the background, and a future request is needed to check the result.

First, note that the first request should return status 202 instead of 200:

The 202 (Accepted) status code indicates that the request has been accepted for processing, but the processing has not been completed. (RFC 9110)

Also, note that you can spawn a new python thread before making the response. This thread can execute the long process, and store the result in a shared variable (or database) accessible by flask.

It is also nice to use the "location" header of the first response (202) to indicate which url should be used later to fetch the result.

Here's the full code:

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()

Note that this code uses the simplest tools that python provides. Measures for thread safety, crash recovery, resource limitation, etc. need to be added to achieve a robust result.