Logging et monitoring FastAPI

Pourquoi Logging et monitoring FastAPI ?

Le logging et le monitoring sont des éléments cruciaux pour la maintenance, l'optimisation et la sécurité d'une application FastAPI. Ils vous aident à identifier rapidement les erreurs, à surveiller les performances de votre application et à comprendre comment elle s'exécute en production.

Un cas d'usage concret est le suivi du comportement des utilisateurs sur une plateforme e-commerce. En analysant les logs, on peut détecter des anomalies ou des comportements inhabituels qui pourraient être des tentatives de fraude. Le monitoring permet ensuite de mettre en place des alertes automatiques pour signaler ces événements.

Prerequis

• Connaissance de base de Python
• Connaissances sur FastAPI (version recommandée : 0.68)
• Installation d'un environnement de développement avec Python et pip
• Un IDE ou un éditeur de code textuel (VSCode, PyCharm)

Concepts fondamentaux

Logging

Le logging est le processus par lequel les informations de diagnostic sont enregistrées dans un fichier ou une autre destination. Cela vous permet de suivre l'exécution de votre application et de déboguer des problèmes.

## 📦 Installation du logger
import logging

## ✅ Configuration du logger
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')

## 👇 Utilisation du logger
logging.debug("Ceci est un message de débogage")
logging.info("Ceci est un message d'information")
logging.warning("Ceci est un avertissement")
logging.error("Ceci est une erreur critique")
logging.critical("Ceci est une erreur critique")

Middleware

Le middleware est une fonction qui reçoit la requête, la traite et passe ensuite à la prochaine partie de l'application. Il peut être utilisé pour ajouter des fonctionnalités comme le logging.

## 📦 Création d'un middleware
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def log_middleware(request: Request, call_next):
    # 👇 Logging de la requête
    logging.info(f"Request: {request.method} {request.url}")
    
    response = await call_next(request)
    
    # 👇 Logging de la réponse
    logging.info(f"Response status: {response.status_code}")
    
    return response

Monitoring

Le monitoring consiste à surveiller les performances et l'état d'une application en temps réel. FastAPI a des extensions comme FastAPI-Monitoring qui facilitent ce processus.

## 📦 Installation de FastAPI-Monitoring
pip install fastapi-monitoring

## ✅ Configuration du monitoring
from fastapi import FastAPI
from fastapi_monitoring.middleware import MonitoringMiddleware

app = FastAPI()

app.add_middleware(MonitoringMiddleware)

Mise en pratique : projet fil rouge

Nous allons construire un simple gestionnaire de tâches basé sur FastAPI. Le système permettra d'ajouter, de mettre à jour et de supprimer des tâches.

## 📦 Création du fichier main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Task(BaseModel):
    id: int
    title: str
    completed: bool = False

tasks = []

@app.post("/tasks/", response_model=Task)
def create_task(task: Task):
    tasks.append(task)
    return task

@app.get("/tasks/{task_id}", response_model=Task)
def read_task(task_id: int):
    for task in tasks:
        if task.id == task_id:
            return task
    raise HTTPException(status_code=404, detail="Task not found")

@app.put("/tasks/{task_id}", response_model=Task)
def update_task(task_id: int, updated_task: Task):
    for i, task in enumerate(tasks):
        if task.id == task_id:
            tasks[i] = updated_task
            return updated_task
    raise HTTPException(status_code=404, detail="Task not found")

@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
    global tasks
    tasks = [task for task in tasks if task.id != task_id]

Étapes de création

1. Création du fichier main.py : Nous définissons une classe Task avec les champs nécessaires et une liste pour stocker nos tâches.

2. Définition des routes :

   • POST /tasks/ : Ajoute une nouvelle tâche.
   • GET /tasks/{task_id} : Renvoie une tâche spécifique.
   • PUT /tasks/{task_id} : Met à jour une tâche existante.
   • DELETE /tasks/{task_id} : Supprime une tâche.

Exécution du serveur

## 🚀 Lancement du serveur FastAPI
uvicorn main:app --reload

Erreurs frequentes et debugging

Erreur 1 : Mauvais format de logging

Code incorrect :

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')

Erreur :

ValueError: Unknown argument: 'format'

Correction :

## 📦 Correction du format de logging
import logging

logging.basicConfig(level=logging.DEBUG, filename='app.log', filemode='w',
                    format='%(asctime)s - %(levelname)s - %(message)s')

Erreur 2 : Middleware non appelé

Code incorrect :

@app.middleware("http")
async def log_middleware(request: Request, call_next):
    logging.info(f"Request: {request.method} {request.url}")
    response = await call_next(request)
    logging.info(f"Response status: {response.status_code}")

Erreur :

AttributeError: module 'fastapi.middleware' has no attribute 'middleware'

Correction :

## 📦 Correction du middleware
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def log_middleware(request: Request, call_next):
    logging.info(f"Request: {request.method} {request.url}")
    response = await call_next(request)
    logging.info(f"Response status: {response.status_code}")
    return response

Erreur 3 : Erreur lors de la suppression d'une tâche

Code incorrect :

@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
    tasks = [task for task in tasks if task.id != task_id]

Erreur :

UnboundLocalError: local variable 'tasks' referenced before assignment

Correction :

## 📦 Correction de la suppression d'une tâche
@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
    global tasks
    tasks = [task for task in tasks if task.id != task_id]

Pour aller plus loin

1. Utilisation de Sentry pour le monitoring des erreurs : https://docs.sentry.io/platforms/python/
2. Intégration avec Prometheus et Grafana pour les métriques avancées : https://www.fastapimvc.org/integrations/prometheus/
3. Ajout de tests unitaires avec Pytest : https://fastapi.tiangolo.com/tutorial/tests/

Défi pratique

Créez une API de blog simple qui permet d'ajouter, lire, mettre à jour et supprimer des articles. Utilisez le middleware pour enregistrer les requêtes et le logging pour suivre l'exécution de votre application.

N'hésitez pas à me poser des questions ou à partager vos commentaires !