Comment intégrer MongoDB avec FastAPI pour la gestion de données NoSQL et la validation efficace des données

L'intégration de MongoDB dans une application FastAPI permet de tirer parti des avantages d'une base de données NoSQL, tout en garantissant une gestion dynamique des données, sans nécessiter de schéma rigide. Cela permet aux développeurs de construire des applications web flexibles et évolutives, adaptées aux besoins modernes des utilisateurs. Cette approche, qui repose sur la connexion et la gestion des documents dans MongoDB, inclut également l’utilisation de Pydantic pour la validation et la sérialisation des données.

Lorsque l’on démarre avec MongoDB, il est essentiel de configurer une connexion adéquate à la base de données avant d’intégrer FastAPI pour les opérations de lecture et d’écriture. Une fois la connexion établie, l'ajout de nouveaux utilisateurs ou la récupération d’un utilisateur spécifique devient un processus relativement simple. Dans ce contexte, la méthode insert_one permet d’ajouter un document dans une collection MongoDB, tandis que find_one est utilisée pour récupérer un utilisateur à partir de son identifiant.

Pour ajouter un utilisateur, le code suivant montre comment intégrer un modèle d’utilisateur dans le serveur FastAPI avec une API POST :

python
Copy code
@app.post("/user")

def create_user(user: User):
    result = user_collection.insert_one(
        user.model_dump(exclude_none=True)
    )
    user_response = UserResponse(
        id=str(result.inserted_id),
        *user.model_dump()
    )
    return user_response

Dans cet exemple, un nouvel utilisateur est ajouté à la collection et son identifiant généré est renvoyé dans la réponse. De même, pour récupérer un utilisateur en fonction de son ID, on utilise la méthode find_one de MongoDB en combinant l’ID avec l’objet ObjectId de BSON, car MongoDB utilise un identifiant unique de 12 octets pour chaque document.

python
Copy code
@app.get("/user")

def get_user(user_id: str):
    db_user = user_collection.find_one(
        {"_id": ObjectId(user_id)} if ObjectId.is_valid(user_id) else None
    )
    if db_user is None:
        raise HTTPException(
            status_code=404, detail="User not found"
        )
    user_response = UserResponse(
        id=str(db_user["_id"]), **db_user
    )
    return user_response

En complément, l'intégration de MongoDB dans FastAPI permet de manipuler facilement des données sans schéma strict. Ce modèle NoSQL est particulièrement utile lorsque les données sont susceptibles d'évoluer au fil du temps, ou lorsqu'elles ne suivent pas un format préétabli rigide. Cela s'avère être un atout pour les applications modernes où les exigences changent rapidement.

Une autre facette essentielle dans le développement d'applications robustes avec FastAPI réside dans la validation des données entrantes. FastAPI utilise Pydantic pour valider et sérialiser automatiquement les données JSON envoyées par les utilisateurs. Pydantic repose sur des annotations de types Python, ce qui permet de valider les données à la volée. Cela évite les erreurs courantes d'entrée et garantit que toutes les informations traitées respectent les formats attendus.

python
Copy code
class UserCreate(BaseModel):
    name: str
    email: EmailStr

Lorsque le modèle Pydantic est utilisé dans FastAPI, la validation est automatique : dès qu'un utilisateur soumet une requête POST avec des informations invalides, comme un email mal formé, une réponse détaillée est renvoyée, indiquant l’erreur spécifique :

json
Copy code
{

  "detail": [
    {
      "loc": ["body", "email"],
      "msg": "value is not a valid email address",
      "type": "type_error.email"
    }
  ]
}

Outre la validation simple, FastAPI et Pydantic offrent également des techniques de validation avancées. L'ajout de validateurs personnalisés dans un modèle permet de définir des logiques de validation plus complexes, comme la validation de l’âge d'un utilisateur. Par exemple, on peut restreindre l'âge d'un utilisateur à une plage spécifique, garantissant ainsi que seules les valeurs correctes sont traitées.

python
Copy code
class User(BaseModel):

    name: str
    email: EmailStr
    age: int
    @field_validator("age")
    def validate_age(cls, value):
        if value < 18 or value > 100:
            raise ValueError("Age must be between 18 and 100")
        return value

Cette approche garantit une plus grande sécurité dans les interactions avec les utilisateurs, en assurant que les données respectent non seulement un format spécifique, mais aussi des règles de logique métier précises.

En outre, FastAPI prend en charge la sérialisation et la désérialisation automatiques des données, facilitant l'échange d'informations complexes sous des formats plus simples comme JSON. Lorsqu'un modèle Pydantic est renvoyé par un point d'API, il est automatiquement transformé en JSON. Inversement, lorsque FastAPI reçoit des données JSON, celles-ci sont automatiquement converties en modèles Python correspondants, prêts à être utilisés dans les opérations de l’application.

Comment connecter FastAPI à MongoDB et effectuer des opérations CRUD

Lorsque vous travaillez avec FastAPI et MongoDB, la première étape consiste à établir une connexion avec la base de données. Pour cela, il est essentiel de configurer MongoDB sur votre machine locale ou sur un serveur distant. Vous pouvez consulter la documentation officielle pour obtenir les informations nécessaires à l'installation de MongoDB dans votre environnement : https://www.mongodb.com/try/download/community. Assurez-vous que MongoDB est bien en cours d'exécution et accessible via l'URL http://localhost:27017 ou une autre adresse si vous utilisez un autre port.

Configuration de la connexion MongoDB

Nous commençons par créer un dossier principal pour notre projet, appelé streaming_platform, et un sous-dossier app dans lequel nous définirons un module db_connection.py. Ce fichier contiendra la configuration nécessaire pour établir la connexion avec MongoDB.

Tout d'abord, nous importons le client asynchrone MongoDB à partir de la bibliothèque motor :

python
Copy code
from motor.motor_asyncio import AsyncIOMotorClient

mongo_client = AsyncIOMotorClient("mongodb://localhost:27017")

Nous utiliserons ensuite l'objet mongo_client pour interagir avec la base de données. Il est crucial de vérifier que la connexion à MongoDB fonctionne correctement. Pour ce faire, nous créons une fonction ping_mongo_db_server() dans db_connection.py qui enverra une commande "ping" à la base de données pour s'assurer qu'elle répond correctement.

python
Copy code
import logging

logger = logging.getLogger("uvicorn.error")
async def ping_mongo_db_server():
    try:
        await mongo_client.admin.command("ping")
        logger.info("Connected to MongoDB")
    except Exception as e:
        logger.error(f"Error connecting to MongoDB: {e}")
        raise e

Cette fonction vérifie la connexion au serveur MongoDB et, en cas de problème, elle génère une erreur qui empêchera le démarrage de l'application.

Initialisation du serveur FastAPI

Ensuite, pour que cette vérification de la connexion soit effectuée lors du démarrage de notre serveur FastAPI, nous devons configurer un gestionnaire de contexte lifespan dans le fichier main.py. Ce gestionnaire sera responsable de l'exécution des tâches au début et à la fin du cycle de vie du serveur.

python
Copy code
from contextlib import asynccontextmanager

from app.db_connection import ping_mongo_db_server
@asynccontextmanager
async def lifespan(app: FastAPI):
    await ping_mongo_db_server()
    yield

Nous intégrons ensuite ce gestionnaire dans l'application FastAPI comme suit :

python
Copy code
from fastapi import FastAPI
app = FastAPI(lifespan=lifespan)

Cette configuration garantit que la vérification de la connexion à MongoDB sera effectuée à chaque démarrage du serveur.

Création des opérations CRUD

python
Copy code
from app.db_connection import mongo_client
database = mongo_client.beat_streaming

MongoDB crée automatiquement la base de données beat_streaming si elle n'existe pas. Nous définissons également une fonction pour récupérer cette base de données en tant que dépendance, ce qui facilitera la réutilisation du code dans nos endpoints.

python
Copy code
def mongo_database():

    return database

Ajouter, lire, mettre à jour et supprimer des chansons

Les opérations CRUD pour notre plateforme de streaming sont définies à travers plusieurs endpoints FastAPI.

1. Ajouter une chanson : Le premier endpoint permet d'ajouter une chanson à la collection songs dans MongoDB. Nous utilisons un format JSON pour transmettre les données de la chanson.

python
Copy code
from bson import ObjectId
from fastapi import Body, Depends
from app.database import mongo_database
from fastapi.encoders import ENCODERS_BY_TYPE
ENCODERS_BY_TYPE[ObjectId] = str
@app.post("/song")
async def add_song(song: dict = Body(...), mongo_db=Depends(mongo_database)):
    await mongo_db.songs.insert_one(song)

    return {"message": "Song added successfully", "id": song["_id"]}

2. Lire une chanson : Le deuxième endpoint permet de récupérer une chanson en fonction de son ID. Si l'ID est invalide ou si la chanson n'existe pas, un message d'erreur 404 est renvoyé.

python
Copy code
from fastapi import HTTPException
@app.get("/song/{song_id}")

async def get_song(song_id: str, db=Depends(mongo_database)):

    song = await db.songs.find_one({"_id": ObjectId(song_id) if ObjectId.is_valid(song_id) else None})
    if not song:
        raise HTTPException(status_code=404, detail="Song not found")
    return song

3. Mettre à jour une chanson : Le troisième endpoint permet de mettre à jour les informations d'une chanson en fonction de son ID.

python
Copy code
@app.put("/song/{song_id}")

async def update_song(song_id: str, song: dict, db=Depends(mongo_database)):
    updated_song = await db.songs.find_one_and_update(
        {"_id": ObjectId(song_id)}, {"$set": song}, return_document=True
    )
    if not updated_song:
        raise HTTPException(status_code=404, detail="Song not found")
    return {"message": "Song updated successfully", "song": updated_song}

4. Supprimer une chanson : Le quatrième endpoint permet de supprimer une chanson de la base de données en utilisant son ID.

python
Copy code
@app.delete("/song/{song_id}")

async def delete_song(song_id: str, db=Depends(mongo_database)):
    result = await db.songs.delete_one({"_id": ObjectId(song_id)})
    if result.deleted_count == 0:
        raise HTTPException(status_code=404, detail="Song not found")
    return {"message": "Song deleted successfully"}

Conclusion

L'intégration de MongoDB avec FastAPI permet de créer une plateforme de streaming souple et évolutive. Grâce à la nature asynchrone de FastAPI et du driver motor, les opérations CRUD sont réalisées de manière efficace, sans bloquer les autres processus. La flexibilité de MongoDB en tant que base de données NoSQL permet d'ajouter, de modifier et de supprimer des documents avec une grande souplesse. Cependant, il est important de toujours bien gérer les connexions et les erreurs, afin de garantir que l'application fonctionne de manière stable.

Comment développer des middleware ASGI pour la modification des requêtes et des réponses dans FastAPI

Le middleware est un outil puissant pour manipuler les requêtes et les réponses dans les applications web. Il permet d'intercepter ces messages pour y ajouter ou modifier des fonctionnalités spécifiques. Dans cet article, nous allons explorer deux types principaux de middleware : celui pour la modification des requêtes et celui pour la modification des réponses. Nous nous concentrerons sur l'implémentation d'un middleware ASGI personnalisé dans une application FastAPI, avec une attention particulière à la modification du contenu des corps des requêtes ainsi que l'ajout d'en-têtes personnalisés aux réponses.

Le concept de middleware dans FastAPI repose sur la bibliothèque Starlette, qui permet une manipulation fine du flux de données entre le client et l'application. Le middleware ASGI est particulièrement utile dans le cas où il faut intervenir avant que la requête n'atteigne le gestionnaire d'API ou avant que la réponse ne soit envoyée au client. Cela permet d'ajouter des couches de sécurité, de transformation ou de journalisation.

Modification du corps des requêtes avec un middleware personnalisé

Tout d'abord, assurez-vous d'avoir créé un middleware personnalisé comme indiqué dans les instructions précédentes. Pour ce faire, vous devez d'abord définir un point d'API dans FastAPI qui recevra des données par POST. Un exemple basique de ce point est le suivant :

python
Copy code
import logging
logger = logging.getLogger("uvicorn")
@app.post("/send")

async def send(message: str = Body()):

    logger.info(f"Message: {message}")
    return message

Ce point d'API affichera simplement le contenu du corps de la requête et le renverra dans la réponse. Cependant, nous souhaitons ajouter un middleware pour hacher ce contenu avant qu'il ne parvienne à l'API. Voici comment procéder.

1. Dans le module request_middleware.py, créez une classe HashBodyContentMiddleware qui interceptera les requêtes. Ce middleware prendra une liste de chemins (par exemple /send) et ne s'appliquera qu'à ceux-ci. Le corps des requêtes sera haché à l'aide de la fonction sha1.

python
Copy code
from starlette.types import ASGIApp, Scope, Receive, Send, Message

from hashlib import sha1
class HashBodyContentMiddleware:
    def __init__(self, app: ASGIApp, allowed_paths: list[str]):
        self.app = app
        self.allowed_paths = allowed_paths

    async def __call__(self, scope: Scope, receive: Receive, send: Send):

        if scope["type"] != "http" or scope["path"] not in self.allowed_paths:
            await self.app(scope, receive, send)
            return
        async def receive_with_new_body() -> Message:
            message = await receive()
            assert message["type"] == "http.request"
            body = message.get("body", b"")
            message["body"] = f'"{sha1(body).hexdigest()}"'.encode()
            return message
        await self.app(scope, receive_with_new_body, send)

2. Ce middleware sera ensuite ajouté à l'application FastAPI via la méthode add_middleware, comme suit :

python
Copy code
app.add_middleware(HashBodyContentMiddleware, allowed_paths=["/send"])

Ainsi, chaque fois qu'une requête sera envoyée à l'endpoint /send, le corps de la requête sera modifié par le middleware avant d'atteindre la fonction send.

Modification des réponses avec un middleware personnalisé

Un autre cas d'utilisation courant du middleware consiste à intercepter les réponses avant qu'elles ne soient renvoyées au client. Cela peut être utile pour ajouter des en-têtes de sécurité, personnaliser les réponses ou enregistrer des informations supplémentaires.

Voici un exemple d'un middleware ASGI qui ajoute un en-tête personnalisé à toutes les réponses de l'application. Pour cela, il suffit de créer un middleware qui intercepte la réponse et y ajoute l'en-tête désiré.

1. Créez un module response_middleware.py où vous définirez le middleware de modification des réponses. Ce middleware ajoutera un en-tête X-Custom-Header à chaque réponse.

python
Copy code
from typing import Sequence

from starlette.middleware.base import BaseHTTPMiddleware
class CustomHeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        response.headers['X-Custom-Header'] = 'Valeur Personnalisée'
        return response

2. Ajoutez ce middleware à votre instance FastAPI :

python
Copy code
app.add_middleware(CustomHeaderMiddleware)

Cela ajoutera l'en-tête X-Custom-Header à toutes les réponses de l'application. Cette approche est idéale pour l'ajout de contrôles personnalisés sur chaque réponse, tels que des en-têtes de sécurité ou de traçabilité.

Conclusion

L'utilisation de middleware dans FastAPI permet de traiter de manière efficace et centralisée les requêtes et les réponses. Que ce soit pour modifier le contenu des requêtes avant qu'elles ne parviennent à un gestionnaire, ou pour personnaliser les réponses avant qu'elles ne soient renvoyées au client, les middleware ASGI offrent une flexibilité énorme pour gérer des scénarios complexes.

Les applications pratiques de cette technologie sont vastes : ajout de sécurité, amélioration de la performance par la gestion des sessions, personnalisation des réponses ou encore transformation des données. Il est essentiel de bien comprendre que chaque middleware peut influencer le flux des données et potentiellement introduire des problèmes si mal configuré. Par conséquent, la gestion de la chaîne des middleware doit être soigneusement pensée pour garantir la performance et la sécurité de l'application.

Comment implémenter des webhooks et sécuriser les hôtes dans FastAPI ?

Le développement moderne d'applications web nécessite souvent l'intégration de services tiers et l'interaction en temps réel entre différentes plateformes. Parmi les outils qui facilitent cette interaction, les webhooks jouent un rôle crucial. Un webhook est essentiellement une notification envoyée par un serveur à un autre, souvent en réponse à un événement particulier. Dans le cadre de cette discussion, nous allons explorer l'implémentation de webhooks dans une application FastAPI, ainsi que la manière de sécuriser l'accès à cette application via le middleware TrustedHostMiddleware.

Lorsque nous développons une API, il est fréquent d’avoir besoin de notifier d'autres systèmes ou applications chaque fois qu'un événement particulier se produit dans notre API. Cela peut inclure des notifications de changements de données ou des alertes pour des actions spécifiques. Les webhooks, utilisés comme des callbacks HTTP, permettent de communiquer ces événements à d'autres systèmes en envoyant une requête HTTP à une URL spécifique.

Mise en place de la sécurité avec le middleware TrustedHostMiddleware

Avant de nous plonger dans la création d'un webhook, il est essentiel de comprendre comment sécuriser l'accès à notre API. FastAPI propose un middleware appelé TrustedHostMiddleware, qui permet de limiter les hôtes autorisés à interagir avec votre application. Il est crucial de ne pas laisser votre API exposée à des hôtes non fiables. Cela peut aider à prévenir des attaques comme l'empoisonnement de l'en-tête "Host", où un attaquant pourrait manipuler l'en-tête d'une requête pour rediriger ou intercepter le trafic.

python
Copy code
from fastapi.middleware.trustedhost import TrustedHostMiddleware
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["localhost"],
)

Créer un webhook avec FastAPI

Les webhooks sont de plus en plus utilisés pour créer des applications réactives et intégrées, permettant une communication en temps réel avec d'autres services. Pour implémenter un webhook dans FastAPI, plusieurs étapes sont nécessaires, notamment la configuration de l'enregistrement des URLs, la gestion des callbacks, et l'envoi d'événements aux URLs enregistrées.

Enregistrement des URLs des webhooks

Voici un exemple de code pour enregistrer une URL de webhook :

python
Copy code
from fastapi import FastAPI, Request, Body
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
    yield {"webhook_urls": []}
app = FastAPI(lifespan=lifespan)
@app.post("/register-webhook-url")

async def add_webhook_url(request: Request, url: str = Body()):

    if not url.startswith("http"):
        url = f"http://{url}"
    request.state.webhook_urls.append(url)
    return {"url added": url}

Cet endpoint accepte une URL sous forme de chaîne de caractères, valide qu'elle commence par http, et l'ajoute à la liste des URLs enregistrées. Ce processus garantit que les événements peuvent être envoyés à ces adresses lors de l'occurrence de certains événements.

Mise en place des callbacks de webhook

Le middleware doit capturer chaque requête entrante et envoyer les données pertinentes sous forme d'événements à tous les abonnés enregistrés. Voici comment cela peut être implémenté :

python
Copy code
from pydantic import BaseModel
import logging
from httpx import AsyncClient
from asyncio import create_task
from datetime import datetime
from fastapi import Request
from starlette.types import ASGIApp, Receive, Scope, Send
client = AsyncClient()
logger = logging.getLogger("uvicorn")
class Event(BaseModel):
    host: str
    path: str
    time: str
    body: str | None = None

async def send_event_to_url(url: str, event: Event):

    logger.info(f"Sending event to {url}")
    try:
        await client.post(f"{url}/fastapi-webhook", json=event.dict())
    except Exception as e:
        logger.error(f"Error sending webhook event to {url}: {e}")
class WebhookSenderMiddleware:
    def __init__(self, app: ASGIApp):
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send):

        if scope["type"] == "http":
            message = await receive()
            body = message.get("body", b"")
            request = Request(scope=scope)
            event = Event(
                host=request.client.host,
                path=request.url.path,
                time=datetime.now().isoformat(),
                body=body,
            )
            for url in request.state.webhook_urls:
                await create_task(send_event_to_url(url, event))
        async def continue_receive():
            return message
        await self.app(scope, continue_receive, send)

Ce middleware intercepte toutes les requêtes HTTP, crée un événement avec les données pertinentes, et envoie cet événement à toutes les URLs enregistrées. Cela permet d'assurer que chaque appel à l'API notifie en temps réel tous les abonnés du webhook.

Points importants à comprendre

Lors de l'implémentation de webhooks, il est crucial de comprendre que chaque requête peut entraîner l'envoi d'événements à plusieurs abonnés. Cela peut générer une charge importante sur le serveur si les abonnés sont nombreux ou si les événements sont envoyés fréquemment. Il est donc essentiel de gérer les erreurs efficacement et de s'assurer que le système est capable de gérer un grand volume de notifications sans entraîner de délais importants.

De plus, lors de l'enregistrement des URLs de webhooks, il est impératif de valider et de sécuriser ces URLs afin d'éviter des attaques malveillantes. L'authentification et l'autorisation des URLs des abonnés sont des éléments essentiels pour maintenir la sécurité de l'application. Enfin, bien que l'approche de stockage dans l'état de l'application soit pratique pour des démonstrations et des prototypes, une solution plus robuste impliquant une base de données devrait être envisagée pour des projets à plus grande échelle.