Nouveau : Datasets open source gratuits disponibles !Decouvrir →
sharing-data.fr sharing-data.fr
🐍
Intermediaire 30 min • FastAPI

FastAPI avec MongoDB : guide pratique

Voici un tutoriel approfondi en français : "FastAPI avec MongoDB : guide pratique".

Pourquoi FastAPI avec MongoDB : guide pratique ?

Dans notre monde numérique rapide et dynamique, la performance et l'efficacité sont de plus en plus importantes. C'est là que FastAPI et MongoDB entrent en jeu. FastAPI est une bibliothèque Python pour créer des API web modernes et performantes, tandis que MongoDB est un système de base de données NoSQL flexible et évolutif. Ensemble, ils offrent une solution complète et puissante pour le développement d'applications web.

Un cas concret serait l'application d'un gestionnaire de tâches où il est important de gérer efficacement les données dans un environnement à haute disponibilité et à grande échelle.

Prerequis

  • Connaissances en Python 3.6+
  • Connaissance de FastAPI
  • Connaissance de MongoDB (ou NoSQL en général)
  • Installations nécessaires :
    • Python 3.10
    • pip
    • MongoDB Community Edition

Concepts fondamentaux

FastAPI

FastAPI est une bibliothèque moderne pour créer des API web avec Python 3.7+ basée sur le framework Starlette et les types de données Pydantic. Elle permet de développer des APIs rapides et efficaces en utilisant un style d'écriture similaire à celui des classes.

## Import FastAPI et Pydantic
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

## Modèle de données avec Pydantic
class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

## Route pour créer un item
@app.post("/items/")
async def create_item(item: Item):
    return {"item": item}

MongoDB

MongoDB est une base de données NoSQL qui utilise le langage JSON pour stocker et accéder aux données. Il offre une structure flexible et des performances élevées.

## Importer pymongo pour connecter à MongoDB
from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
db = client["mydatabase"]
collection = db["items"]

## Insertion d'un document dans la collection
item = {"name": "Laptop", "price": 999.99}
result = collection.insert_one(item)
print(f"Inserted item with ID {result.inserted_id}")

Mise en pratique : projet fil rouge

Projet : Gestionnaire de tâches

Nous allons créer un simple gestionnaire de tâches qui permet d'ajouter, récupérer et supprimer des tâches. Cette application sera construite avec FastAPI pour le backend et MongoDB pour le stockage.

Étape 1 : Créer la structure du projet

mkdir task_manager
cd task_manager
mkdir app models schemas tests
touch app/main.py app/db.py models/task_model.py schemas/task_schema.py

Étape 2 : Installer les dépendances

pip install fastapi uvicorn pymongo pydantic[extra]

Étape 3 : Configurer la base de données MongoDB

## app/db.py
from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
db = client["task_manager"]
collection = db["tasks"]

Étape 4 : Définir le modèle de tâche

## models/task_model.py
from pydantic import BaseModel, Field

class Task(BaseModel):
    id: str = Field(None)
    title: str
    description: str
    completed: bool = False

Étape 5 : Définir les schémas de validation

## schemas/task_schema.py
from pydantic import BaseModel, validator
from models.task_model import Task

class TaskCreate(BaseModel):
    title: str
    description: str = None
    completed: bool = False

    @validator('title', pre=True)
    def strip_title(cls, value):
        return value.strip()

class TaskUpdate(BaseModel):
    title: str = None
    description: str = None
    completed: bool = None

Étape 6 : Créer les routes API

## app/main.py
from fastapi import FastAPI, HTTPException
from pymongo.errors import DuplicateKeyError
from schemas.task_schema import TaskCreate, TaskUpdate
from models.task_model import Task
from db import collection

app = FastAPI()

@app.post("/tasks/", response_model=Task)
async def create_task(task_create: TaskCreate):
    task_dict = task_create.dict()
    try:
        result = collection.insert_one(task_dict)
        return {**task_dict, "id": str(result.inserted_id)}
    except DuplicateKeyError:
        raise HTTPException(status_code=409, detail="Task already exists")

@app.get("/tasks/{task_id}", response_model=Task)
async def read_task(task_id: str):
    task = collection.find_one({"_id": task_id})
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return {**task, "id": str(task["_id"])}

@app.put("/tasks/{task_id}", response_model=Task)
async def update_task(task_id: str, task_update: TaskUpdate):
    task_dict = task_update.dict(exclude_unset=True)
    result = collection.update_one({"_id": task_id}, {"$set": task_dict})
    if not result.modified_count:
        raise HTTPException(status_code=404, detail="Task not found")
    updated_task = collection.find_one({"_id": task_id})
    return {**updated_task, "id": str(updated_task["_id"])}

@app.delete("/tasks/{task_id}", response_model=dict)
async def delete_task(task_id: str):
    result = collection.delete_one({"_id": task_id})
    if not result.deleted_count:
        raise HTTPException(status_code=404, detail="Task not found")
    return {"message": "Task deleted"}

Étape 7 : Lancer l'application

uvicorn app.main:app --reload

Erreurs frequentes et debugging

Erreur 1 : MongoDB ne trouve pas la collection

Code incorrect :

## db.py
from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
db = client["mydatabase"]
collection = db["items"]

Code correct :

## db.py
from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
db = client["task_manager"]
collection = db["tasks"]

Erreur 2 : FastAPI renvoie une erreur de validation

Code incorrect :

## schemas/task_schema.py
from pydantic import BaseModel, validator
from models.task_model import Task

class TaskCreate(BaseModel):
    title: str
    description: str = None
    completed: bool = False

    @validator('title', pre=True)
    def strip_title(cls, value):
        return value.strip()

class TaskUpdate(BaseModel):
    title: str = None
    description: str = None
    completed: bool = None

Code correct :

## schemas/task_schema.py
from pydantic import BaseModel, validator
from models.task_model import Task

class TaskCreate(BaseModel):
    title: str
    description: str = None
    completed: bool = False

    @validator('title', pre=True)
    def strip_title(cls, value):
        return value.strip() if value else None

class TaskUpdate(BaseModel):
    title: str = None
    description: str = None
    completed: bool = None

Erreur 3 : MongoDB ne trouve pas le document à mettre à jour

Code incorrect :

## app/main.py
from fastapi import FastAPI, HTTPException
from pymongo.errors import DuplicateKeyError
from schemas.task_schema import TaskCreate, TaskUpdate
from models.task_model import Task
from db import collection

app = FastAPI()

@app.put("/tasks/{task_id}", response_model=Task)
async def update_task(task_id: str, task_update: TaskUpdate):
    task_dict = task_update.dict(exclude_unset=True)
    result = collection.update_one({"_id": task_id}, {"$set": task_dict})
    if not result.modified_count:
        raise HTTPException(status_code=404, detail="Task not found")
    updated_task = collection.find_one({"_id": task_id})
    return {**updated_task, "id": str(updated_task["_id"])}

Code correct :

## app/main.py
from fastapi import FastAPI, HTTPException
from pymongo.errors import DuplicateKeyError
from schemas.task_schema import TaskCreate, TaskUpdate
from models.task_model import Task
from db import collection

app = FastAPI()

@app.put("/tasks/{task_id}", response_model=Task)
async def update_task(task_id: str, task_update: TaskUpdate):
    task_dict = task_update.dict(exclude_unset=True)
    result = collection.update_one({"_id": task_id}, {"$set": task_dict})
    if not result.modified_count:
        raise HTTPException(status_code=404, detail="Task not found")
    updated_task = collection.find_one({"_id": task_id})
    return {**updated_task, "id": str(updated_task["_id"])}

Pour aller plus loin

Piste 1 : Ajouter une authentification

FastAPI offre des fonctionnalités intégrées pour l'authentification. Vous pouvez utiliser des OAuth2 et JWT pour sécuriser votre API.

Documentation FastAPI sur l'authentification

Piste 2 : Utiliser MongoDB Atlas

MongoDB Atlas est une plateforme cloud qui vous permet de héberger vos bases de données MongoDB en ligne. Cela offre des avantages tels que la haute disponibilité, les mises à jour automatiques et le support du clustering.

Documentation MongoDB Atlas

Piste 3 : Créer une application mobile

Vous pouvez créer une application mobile qui consomme votre API FastAPI avec MongoDB pour gérer les tâches. Vous pouvez utiliser Flutter ou React Native pour développer l'application mobile.

Documentation Flutter Documentation React Native

Défi pratique

Créez une application de gestion de contacts en utilisant FastAPI et MongoDB. L'application doit permettre d'ajouter, récupérer, mettre à jour et supprimer des contacts.

Ce tutoriel approfondi devrait vous aider à démarrer un projet avec FastAPI et MongoDB, et vous donner les outils nécessaires pour le développer et le maintenir.

Besoin d'aide sur FastAPI ?

Besoin d'aide sur un projet technique ? Decrivez-le pour des conseils personnalises.

Recevoir des conseils

Questions frequentes

Comment installer FastAPI avec MongoDB dans un environnement Python existant?
Pour installer FastAPI et PyMongo (la bibliothèque Python pour accéder à MongoDB), utilisez pip : `pip install fastapi pymongo`. Ensuite, configurez une connexion à votre base de données MongoDB en utilisant PyMongo.
Comment créer une route GET simple dans un FastAPI qui récupère des données depuis MongoDB?
Pour créer une route GET, utilisez l'annotation `@app.get()`. Exemple : `@app.get('/items') def read_items(): item = db.items.find_one(); return {'item': item}`. Assurez-vous d'avoir configuré la connexion à MongoDB et de sélectionné la collection appropriée.
Comment mettre à jour des données dans MongoDB depuis une route FastAPI?
Pour mettre à jour des données, utilisez la méthode `update_one()` ou `update_many()`. Exemple : `@app.put('/items/{item_id}') def update_item(item_id: str, item_data: Item): result = db.items.update_one({'_id': ObjectId(item_id)}, {'$set': item_data.dict()}); return {'updated_count': result.modified_count}`. Assurez-vous d'avoir le bon type de données et d'effectuer les validations nécessaires.

Pages liees

CheatsheetCheatsheet FastAPIErreursErreurs frequentes FastAPITutorielsTous les tutoriels

Chaque semaine, le meilleur de la tech francaise

Tendances, salaires, outils et opportunites — directement dans votre boite mail.

Gratuit. Desabonnement en un clic. Pas de spam.