Créer une tâche asynchrone dans Flask

J'écris une application dans Flask, qui fonctionne très bien sauf qu'elle WSGIest synchrone et bloquante. J'ai une tâche en particulier qui fait appel à une API tierce et cette tâche peut prendre plusieurs minutes. Je voudrais faire cet appel (c'est en fait une série d'appels) et le laisser fonctionner. tandis que le contrôle est retourné à Flask.

Ma vue ressemble à:

@app.route('/render/<id>', methods=['POST'])
def render_script(id=None):
    ...
    data = json.loads(request.data)
    text_list = data.get('text_list')
    final_file = audio_class.render_audio(data=text_list)
    # do stuff
    return Response(
        mimetype='application/json',
        status=200
    )

Maintenant, ce que je veux faire, c'est avoir la ligne

final_file = audio_class.render_audio()

run et fournissez un rappel à exécuter lorsque la méthode retourne, tandis que Flask peut continuer à traiter les requêtes. C'est la seule tâche dont j'ai besoin que Flask s'exécute de manière asynchrone, et j'aimerais avoir des conseils sur la meilleure façon de l'implémenter.

J'ai regardé Twisted et Klein, mais je ne suis pas sûr qu'ils soient exagérés, car peut-être que Threading suffirait. Ou peut-être que le céleri est un bon choix pour cela?

J'utiliserais Celery pour gérer la tâche asynchrone pour vous. Vous devrez installer un courtier pour servir de file d'attente de tâches (RabbitMQ et Redis sont recommandés).

app.py:

from flask import Flask
from celery import Celery

broker_url = 'amqp://guest@localhost'          # Broker URL for RabbitMQ task queue

app = Flask(__name__)    
celery = Celery(app.name, broker=broker_url)
celery.config_from_object('celeryconfig')      # Your celery configurations in a celeryconfig.py

@celery.task(bind=True)
def some_long_task(self, x, y):
    # Do some long task
    ...

@app.route('/render/<id>', methods=['POST'])
def render_script(id=None):
    ...
    data = json.loads(request.data)
    text_list = data.get('text_list')
    final_file = audio_class.render_audio(data=text_list)
    some_long_task.delay(x, y)                 # Call your async task and pass whatever necessary variables
    return Response(
        mimetype='application/json',
        status=200
    )

Exécutez votre application Flask et démarrez un autre processus pour exécuter votre céleri.

$ celery worker -A app.celery --loglevel=debug

Je ferais également référence à la rédaction de Miguel Gringberg pour un guide plus détaillé sur l'utilisation du céleri avec Flask.

Le filetage est une autre solution possible. Bien que la solution basée sur Celery soit meilleure pour les applications à grande échelle, si vous ne vous attendez pas à trop de trafic sur le point final en question, le threading est une alternative viable.

Cette solution est basée sur la présentation PyCon 2016 Flask at Scale de Miguel Grinberg , en particulier la diapositive 41 de son diaporama. Son code est également disponible sur github pour ceux qui s'intéressent à la source originale.

Du point de vue de l'utilisateur, le code fonctionne comme suit:

1. Vous appelez le point de terminaison qui exécute la tâche de longue durée.
2. Ce point de terminaison renvoie 202 Accepté avec un lien pour vérifier l'état de la tâche.
3. Les appels au lien d'état retournent 202 pendant que le taks est toujours en cours d'exécution, et renvoie 200 (et le résultat) lorsque la tâche est terminée.

Pour convertir un appel d'API en tâche d'arrière-plan, ajoutez simplement le décorateur @async_api.

Voici un exemple complet:

from flask import Flask, g, abort, current_app, request, url_for
from werkzeug.exceptions import HTTPException, InternalServerError
from flask_restful import Resource, Api
from datetime import datetime
from functools import wraps
import threading
import time
import uuid

tasks = {}

app = Flask(__name__)
api = Api(app)


@app.before_first_request
def before_first_request():
    """Start a background thread that cleans up old tasks."""
    def clean_old_tasks():
        """
        This function cleans up old tasks from our in-memory data structure.
        """
        global tasks
        while True:
            # Only keep tasks that are running or that finished less than 5
            # minutes ago.
            five_min_ago = datetime.timestamp(datetime.utcnow()) - 5 * 60
            tasks = {task_id: task for task_id, task in tasks.items()
                     if 'completion_timestamp' not in task or task['completion_timestamp'] > five_min_ago}
            time.sleep(60)

    if not current_app.config['TESTING']:
        thread = threading.Thread(target=clean_old_tasks)
        thread.start()


def async_api(wrapped_function):
    @wraps(wrapped_function)
    def new_function(*args, **kwargs):
        def task_call(flask_app, environ):
            # Create a request context similar to that of the original request
            # so that the task can have access to flask.g, flask.request, etc.
            with flask_app.request_context(environ):
                try:
                    tasks[task_id]['return_value'] = wrapped_function(*args, **kwargs)
                except HTTPException as e:
                    tasks[task_id]['return_value'] = current_app.handle_http_exception(e)
                except Exception as e:
                    # The function raised an exception, so we set a 500 error
                    tasks[task_id]['return_value'] = InternalServerError()
                    if current_app.debug:
                        # We want to find out if something happened so reraise
                        raise
                finally:
                    # We record the time of the response, to help in garbage
                    # collecting old tasks
                    tasks[task_id]['completion_timestamp'] = datetime.timestamp(datetime.utcnow())

                    # close the database session (if any)

        # Assign an id to the asynchronous task
        task_id = uuid.uuid4().hex

        # Record the task, and then launch it
        tasks[task_id] = {'task_thread': threading.Thread(
            target=task_call, args=(current_app._get_current_object(),
                               request.environ))}
        tasks[task_id]['task_thread'].start()

        # Return a 202 response, with a link that the client can use to
        # obtain task status
        print(url_for('gettaskstatus', task_id=task_id))
        return 'accepted', 202, {'Location': url_for('gettaskstatus', task_id=task_id)}
    return new_function


class GetTaskStatus(Resource):
    def get(self, task_id):
        """
        Return status about an asynchronous task. If this request returns a 202
        status code, it means that task hasn't finished yet. Else, the response
        from the task is returned.
        """
        task = tasks.get(task_id)
        if task is None:
            abort(404)
        if 'return_value' not in task:
            return '', 202, {'Location': url_for('gettaskstatus', task_id=task_id)}
        return task['return_value']


class CatchAll(Resource):
    @async_api
    def get(self, path=''):
        # perform some intensive processing
        print("starting processing task, path: '%s'" % path)
        time.sleep(10)
        print("completed processing task, path: '%s'" % path)
        return f'The answer is: {path}'


api.add_resource(CatchAll, '/<path:path>', '/')
api.add_resource(GetTaskStatus, '/status/<task_id>')


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

Vous pouvez également essayer d'utiliser multiprocessing.Processavec daemon=True; la process.start()méthode ne se bloque pas et vous pouvez renvoyer une réponse / un statut immédiatement à l'appelant pendant que votre fonction coûteuse s'exécute en arrière-plan.

J'ai rencontré un problème similaire en travaillant avec le framework Falcon et en utilisant le daemonprocessus aidé.

Vous devez effectuer les opérations suivantes:

from multiprocessing import Process

@app.route('/render/<id>', methods=['POST'])
def render_script(id=None):
    ...
    heavy_process = Process(  # Create a daemonic process with heavy "my_func"
        target=my_func,
        daemon=True
    )
    heavy_process.start()
    return Response(
        mimetype='application/json',
        status=200
    )

# Define some heavy function
def my_func():
    time.sleep(10)
    print("Process finished")

Vous devriez obtenir une réponse immédiatement et, après 10 secondes, vous devriez voir un message imprimé dans la console.

REMARQUE: gardez à l'esprit que les daemonicprocessus ne sont pas autorisés à générer des processus enfants.