Pourquoi Optimiser les performances FastAPI ?
Dans le développement moderne, la performance est une préoccupation majeure pour tout projet, que ce soit un petit script Python ou une application web de grande envergure. Un développeur FastAPI peut avoir besoin d'optimiser les performances à différents moments : au début du développement pour s'assurer qu'on ne construit pas des bottes de performance dès le départ, ou bien plus tard pour améliorer les temps de réponse et l'utilisation des ressources lorsqu'une application commence à monter en charge.
Un cas d'utilisation concret serait une API de blogging avec un grand nombre de requêtes simultanées. Si la performance n'est pas optimisée, on risque d'atteindre le "limite de scale" très rapidement, où l'application ne peut plus gérer les demandes en temps utile. Cela peut entraîner des retards pour les utilisateurs et même une perte de confiance dans le service.
Prerequis
- Connaissance avancée de Python
- Familiarité avec FastAPI
- Connaissance des concepts de gestion des requêtes HTTP, du middleware et de l'injection de dépendances.
- Outils à installer :
- Python (3.7+)
- FastAPI
- Uvicorn ou any ASGI server
Concepts fondamentaux
Middleware
Le middleware est une fonction qui reçoit la requête avant qu'elle n'atteigne le endpoint et la réponse avant de revenir au client. C'est une excellente place pour ajouter des traitements comme la validation, l'enregistrement des logs ou même l'authentification.
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
Dependency Injection
L'injection de dépendances est un pattern qui permet d'injecter des objets nécessaires dans les endpoints. Cela peut être très utile pour partager des ressources comme une base de données, ou pour passer des paramètres communs à plusieurs endpoints.
from fastapi import FastAPI, Depends
app = FastAPI()
async def common_parameters(q: str = None, skip: int = 0, limit: int = 10):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
return commons
Asynchronous Code
FastAPI est fondé sur les fonctions asynchrones, ce qui signifie qu'elles peuvent être exécutées en parallèle. Cela permet d'améliorer la performance pour des opérations I/O comme les appels à une base de données ou un service externe.
import asyncio
@app.get("/items/")
async def read_items():
await asyncio.sleep(1) # Simulate a long-running task
return {"message": "Done"}
Mise en pratique : projet fil rouge
Nous allons construire un simple gestionnaire de tâches asynchrone. Ce projet comprendra les fonctionnalités suivantes :
- Ajouter une tâche
- Obtenir toutes les tâches
- Supprimer une tâche
Étape 1 : Création du Projet
Commencez par créer un nouveau répertoire pour le projet et initialisez un environnement virtuel.
mkdir task_manager
cd task_manager
python -m venv venv
source venv/bin/activate # Sur Windows utilisez `venv\Scripts\activate`
Étape 2 : Installation de FastAPI et Uvicorn
Installez FastAPI et Uvicorn dans votre environnement virtuel.
pip install fastapi uvicorn
Étape 3 : Structure du Projet
Créez les fichiers suivants :
main.py: Le point d'entrée de l'application.models.py: Pour définir la structure des données (ici, une tâche).database.py: Gestionnaire de base de données asynchrone.
task_manager/
├── main.py
├── models.py
└── database.py
Étape 4 : Modèle des Données (models.py)
from pydantic import BaseModel
class Task(BaseModel):
id: int
title: str
description: str = None
completed: bool = False
Étape 5 : Gestionnaire de Base de Données (database.py)
import asyncio
from typing import List, Dict
tasks_db: Dict[int, Task] = {}
async def add_task(task: Task) -> Task:
task.id = len(tasks_db) + 1
tasks_db[task.id] = task
return task
async def get_all_tasks() -> List[Task]:
return list(tasks_db.values())
async def delete_task(task_id: int):
if task_id in tasks_db:
del tasks_db[task_id]
Étape 6 : Point d'Entrée de l'Application (main.py)
from fastapi import FastAPI, HTTPException
from pydantic import ValidationError
import asyncio
app = FastAPI()
from models import Task
from database import add_task, get_all_tasks, delete_task
@app.post("/tasks/")
async def create_task(task: Task):
try:
task = await add_task(task)
return task
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/tasks/")
async def read_tasks():
tasks = await get_all_tasks()
return tasks
@app.delete("/tasks/{task_id}")
async def delete_task_endpoint(task_id: int):
try:
await delete_task(task_id)
return {"message": "Task deleted"}
except KeyError:
raise HTTPException(status_code=404, detail="Task not found")
Étape 7 : Exécution de l'Application
Exécutez l'application avec Uvicorn.
uvicorn main:app --reload
Maintenant, vous pouvez tester votre API en utilisant un outil comme curl ou Postman. Par exemple :
- Ajouter une tâche :
POST /tasks/ - Lister toutes les tâches :
GET /tasks/ - Supprimer une tâche :
DELETE /tasks/{task_id}
Erreurs fréquentes et debugging
1. Depends est exécuté à chaque requête
Code incorrect :
from fastapi import Depends
async def get_db():
# Simulate a database connection
await asyncio.sleep(2)
return "db_connection"
@app.get("/items/")
async def read_items(db=Depends(get_db)):
return {"db": db}
Code correct :
from fastapi import Depends, FastAPI
app = FastAPI()
async def get_db():
# Simulate a database connection
await asyncio.sleep(2)
return "db_connection"
@app.get("/items/")
async def read_items(db: str = Depends(get_db)):
return {"db": db}
2. await manquant
Code incorrect :
import time
@app.get("/long-task")
def long_task():
time.sleep(5) # Synchronous sleep
return {"message": "Done"}
Code correct :
import asyncio
@app.get("/long-task")
async def long_task():
await asyncio.sleep(5) # Asynchronous sleep
return {"message": "Done"}
3. HTTPException non géré
Code incorrect :
from fastapi import HTTPException
@app.get("/items/{item_id}")
def read_item(item_id: int):
if item_id == 42:
raise HTTPException(status_code=404, detail="Item not found")
return {"item_id": item_id}
Code correct :
from fastapi import HTTPException
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id == 42:
raise HTTPException(status_code=404, detail="Item not found")
return {"item_id": item_id}
Pour aller plus loin
- Profiling avec
uvicorn --log-level debug: Utilisez cette option pour obtenir des informations détaillées sur le temps d'exécution de chaque requête. - Utilisation de Redis : Pour stocker les données en mémoire, ce qui peut améliorer significativement la performance.
- Batching : Grouper plusieurs requêtes en une seule pour réduire les coûts.
Défi pratique : Ajoutez une fonctionnalité pour mettre à jour une tâche existante et gérer les erreurs liées à l'existence de cette tâche.
Ce tutoriel a couvert les concepts fondamentaux d'optimisation des performances en FastAPI, ainsi que la mise en pratique d'un projet fil rouge complet. En suivant ces étapes et en résolvant les erreurs fréquentes, vous devriez être en mesure de construire des applications FastAPI performantes à grande échelle.