Manager les tâches en arrière-plan de FastAPI
Aujourd’hui, nous allons voir une implémentation simple qui permet d’exécuter et de suivre des tâches en arrière-plan à l’aide d’une API REST développée avec le Framework FastAPI.
Usage standard d'une API Rest
Le premier usage classique et basique d’une API REST consiste à rediriger le résultat de traitements vers des URL comme dans l’exemple suivant :
from fastapi import FastAPI
app = FastAPI()
def ma_fonction():
return {"message":"Mon super process"}
@app.get("/ma_fonction/")
async def get_ma_fonction():
resultat = ma_fonction()
return resultatCette implémentation est très efficace à la seule condition que la vitesse d’exécution du traitement, celui de la fonction ma_fonction() dans l’exemple précédent, soit très rapide.
En général, cette d’implémentation est utilisée pour retourner des résultats de requêtes effectuées dans une base de données.
Les tâches d'arrière-plan
Toutefois, il n’est pas toujours nécessaire de retourner le résultat d’un traitement.
Dans ce cas, il est souhaitable que le traitement soit effectué en arrière-plan et qu’une réponse indiquant au client la prise en compte de sa requête lui soit retournée par l’API.
Sur le même principe que l’exemple précédent, voici ce que cela donne :
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def ma_fonction():
return {"message":"Mon super process"}
@app.get("/ma_fonction/")
async def get_ma_fonction(background_tasks: BackgroundTasks):
background_tasks.add_task(ma_fonction)
return {"message":"Votre requête a bien été prise en compte"}Monitorer les tâches en arrière-plan
La classe BackgroundTasks est très pratique, car elle permet d’exécuter un traitement en tâche de fond sans être bloquante. Toutefois, elle ne permet pas de lister les traitements en cours d’exécution, ni d’en connaître le contenu.
Pour pouvoir obtenir ses informations, je vous propose d’implémenter une classe dont le rôle va être de manager les tâches.
En premier lieu, le rôle de cette classe va être de référencer, dans une structure de données compatible JSON, les tâches entrantes et de déréférencer les tâches sortantes. Cette pratique est fonctionnelle uniquement parce que par défaut les processus utilisés par FastAPI et par la classe BackgroundTasks autorisent le partage de données et d’objets entre processus.
Reprenons, la première étape consiste à définir un dictionnaire qui doit être consultable par l’API et qui retournera le contenu des traitements en cours.
class TasksManager:
tasks = {}Le choix d’un dictionnaire n’est pas anodin. En effet, la conversion entre dictionnaire et JSON est native dans FastAPI.
Ensuite, nous pouvons ajouter, dans la classe, notre première fonction qui va devoir se référencer dans le dictionnaire avant d’effectuer le traitement principal, puis se déréférencer à la fin de celui-ci.
Pour identifier la tâche comme étant unique, je génère au préalable un UUID qui sera utilisé comme clé dans le dictionnaire. J’ai choisi arbitrairement comme valeur à associer à la clé le nom de la fonction que je récupère à l’aide de __name__.
import uuid
def ma_fonction():
return {"message":"Mon super process"}
class TasksManager:
tasks = {}
def ma_fonction(self):
""" Task which launch ma_fonction() """
# Generate unique UUID to identify task
identifiant = uuid.uuid1()
# Store task name into dictionary
self.tasks[identifiant] = __name__
# Launch function
result = ma_fonction()
# Delete task
del self.tasks[identifiant]
return resultAu niveau de l’API, il faudra créer une instance de TasksManager et remplacer l’appel direct de ma_fonction.
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
tasks_manager = TasksManager()
def ma_fonction(background_tasks: BackgroundTasks):
return {"message":"Mon super process"}
@app.get("/ma_fonction/")
async def get_ma_fonction():
background_tasks.add_task(tasks_manager.ma_fonction)
return {"message":"Votre requête a bien été prise en compte"}Monitorer les tâches en arrière-plan
Il ne reste plus qu’à ajouter, au niveau de l’API, une fonction qui fera le lien entre le dictionnaire tasks de l’instance de la classe TasksManager et une URL.
@app.get("/background-tasks")
async def get_background_tasks():
""" Get list of running background tasks """
background_tasks = tasks_manager.tasks
return background_tasksNous pouvons même créer dynamiquement une URL dédiée à chaque tâche en utilisant leur UUID :
@app.get("/background-tasks/{uuid}")
async def get_background_tasks(uuid: str):
""" Get list of running background tasks """
if uuid in tasks_manager.tasks.keys():
background_task = tasks_manager.tasks[uuid]
return background_task
else :
raise HTTPException(status_code=404, detail="uuid not exist")
background_tasks = tasks_manager.tasks
return background_tasksNous avons maintenant une première implémentation fonctionnelle d’un gestionnaire de tâche en arrière-plan.
Utiliser des décorateurs pour référencer les tâches
Si l’on doit ajouter de nouvelles fonctions dans le gestionnaire de tâche, elles devront toutes respecter le même schéma de pré-traitement et de post-traitement que notre première tâche.
Or, pour palier au problème du copier-coller, on va déléguer la gestion du dictionnaire et de l’uuid à un décorateur privé de la classe.
import uuid
def ma_fonction():
return {"message":"Mon super process"}
class TasksManager:
tasks = {}
def _managed(func):
""" decorator which manage tasks with pre and post process """
def wrapped_func(self, *args, **kwargs):
""" Manage task by adding task args and kwargs inside a dictionary """
# Generate unique UUID to identify task
task_identifier = uuid.uuid1()
task_name = func.__name__
# task registration
self.tasks[task_identifier] = task_name
# launch function
result = func(self, *args, **kwargs)
# delete task
del self.tasks[task_identifier]
return result
return wrapped_func
@_managed
def ma_fonction(self):
""" Task which launch ma_fonction() """
result = ma_fonction()
return resultGrâce au décorateur @_managed, l’ajout de fonction dans le gestionnaire de tâche ne nécessite rien d’autre que de précéder la fonction du décorateur.
Attention, vous noterez que le décorateur ne possède pas de self mais qu’il est parfaitement capable d’accéder aux propriétés de la classe et de fonctionner avec les méthodes de celle-ci.
Notez aussi que le nom du décorateur étant précédé de _, celui-ci est privé à la classe.
Bien Gérer les arguments et les exceptions
Jusqu’ici, la fonction de l’exemple est simple. Mais le rôle d’une gestionnaire de tâche n’est pas uniquement de monitorer des tâches, il sert aussi à gérer les exceptions pour n’en passer aucune silencieusement et il doit aussi s’adapter aux différents arguments de vos fonctions.
Pour cela, il nous suffit juste d’adapter le code dans notre décorateur.
La gestion des exceptions est faite à l’aide de try et except qui viennent englober l’appel de la méthode :
# launch function
try:
result = func(self)
except Exception as e:
# Manage error Here
print("Report an Error")
result = NoneEnfin, la récupération des arguments de la fonction se fait à l’aide de *args et **kwargs qui permettent de récupérer tous les arguments et arguments à mots clés de la fonction décorée.
def wrapped_func(self, *args, **kwargs):
# launch function
try:
result = func(self, *args, **kwargs)
except Exception as e:
# Manage error Here
print("Report an Error")
result = NoneAjouter les *args et **kwargs dans un dictionnaire
Pour finir, notre dictionnaire est pour le moment assez pauvre puisse qu’il ne retourne que le nom de la fonction. Nous allons donc l’enrichir des *args et **kwargs.
Concernant les arguments, il faudra les convertir en dictionnaire pour pouvoir les ajouter proprement à tasks.
# Je récupère le nom de la fonction
task_name = func.__name__
# Je récupère le nom des arguments à l'aide du module inspect
args_name = inspect.getargspec(func)[0]
# Je supprime self pour pouvoir zipper les autres arguments
args_name.remove("self")
# Je zip la liste d'arguments avec la liste des noms d'arguments.
task_dict_args = dict(zip(args_name, args))
# Je fais une copie des kwargs à l'aide de dict()
# task_dict ne pourra jamais être altéré tandis que **kwargs est susceptible d'être altéré.
task_dict_kwargs = dict(kwargs)
# Enregistrement de la tâche avec son nom, ses args et kwargs
self.tasks[task_identifier] = {"name":task_name,
"args": task_dict_args,
"kwargs": task_dict_kwargs}Voilà, nous avons maintenant un joli gestionnaire de tâche complet.
Je finirai simplement en précisant que pour une meilleure organisation, le projet est organisé de la façon suivante :
- api.py
- tasks.py
- fonctions.py
Code complet
Si vous vous êtes perdu en cours de route, pas de panique. Vous trouverez ci-dessous le code complet.
api.py
from fastapi import FastAPI, BackgroundTasks, HTTPException
from tasks import TasksManager()
app = FastAPI()
tasks_manager = TasksManager()
def ma_fonction(background_tasks: BackgroundTasks):
return {"message":"Mon super process"}
@app.get("/ma_fonction/")
async def get_ma_fonction():
background_tasks.add_task(tasks_manager.ma_fonction)
return {"message":"Votre requête a bien été prise en compte"}
@app.get("/background-tasks")
async def get_background_tasks():
""" Get list of running background tasks """
background_tasks = tasks_manager.tasks
return background_tasks
@app.get("/background-tasks/{uuid}")
async def get_background_tasks(uuid: str):
""" Get list of running background tasks """
if uuid in tasks_manager.tasks.keys():
background_task = tasks_manager.tasks[uuid]
return background_task
else :
raise HTTPException(status_code=404, detail="uuid not exist")
background_tasks = tasks_manager.tasks
return background_taskstasks.py
import uuid
import inspect
from fonctions import ma_fonction
class TasksManager:
tasks = {}
def _managed(func):
""" decorator which manage tasks with pre and post process """
def wrapped_func(self, *args, **kwargs):
""" Manage task by adding task args and kwargs inside a dictionary """
# Generate unique UUID to identify task
task_identifier = uuid.uuid1()
task_name = func.__name__
# retrieve function args and kwargs
args_name = inspect.getargspec(func)[0]
args_name.remove("self")
task_dict_args = dict(zip(args_name, args))
task_dict_kwargs = dict(kwargs)
# task registration
self.tasks[task_identifier] = {"name":task_name,
"args": task_dict_args,
"kwargs": task_dict_kwargs}
# launch function
try:
result = func(self, *args, **kwargs)
except Exception as e:
# Manage error Here
print("Report an Error")
result = None
# delete task
del self.tasks[task_identifier]
return result
return wrapped_func
@_managed
def ma_fonction(self):
""" Task which launch ma_fonction() """
result = ma_fonction()
return resultfonctions.py
def ma_fonction():
return {"message":"Mon super process"}Le mot de la fin
L’implémentation que je vous aie proposé est très naïve et a surtout pour objectif de vous apporter des éléments de réflexion sur ce que vous pouvez faire pour ajouter des fonctionnalités de suivi de process à votre API sans pour autant avoir besoin d’une librairie externe tel que Celery, RedisQueue, …
En réalité, si je devais récrire le truc, je me débrouillerais pour simplement avoir un décorateur à appliquer sur n’importe quelle fonction sans devoir la référencer dans la classe du manager et qui la rendrait visible via l’API. Si un jour, j’ai le temps, je repasserai sur cet article pour implémenter cette façon de faire qui est plus pragmatique.
Dans tous les cas, j’espère que cet article t’aura plu, qu’il t’aura appris des trucs et potentiellement qu’il t’a été utile ! Si tu veux me faire part de quelque chose, n’hésitez pas à m’écrire dans les commentaires ou via la page de contact.
