Comment installer Fast API avec MongoDB sur Ubuntu 24.04

FastAPI est un framework web basé sur Python pour créer des services API. C’est un framework moderne, rapide et performant qui prend en charge les opérations asynchrones.

Dans ce tutoriel, vous apprendrez comment installer FastAPI avec MongoDB sur Ubuntu 24.04. Vous apprendrez également comment créer votre première API avec des opérations CRUD en utilisant FastAPI et la base de données MongoDB.

Prérequis

Pour commencer ce guide, assurez-vous d’avoir les éléments suivants :

• Un système Ubuntu 24.04
• Un utilisateur non-root avec des privilèges d’administrateur

Installation de MongoDB

Avant de créer un nouveau projet FastAPI, installons le serveur MongoDB sur notre système.

Tout d’abord, exécutez la commande ci-dessous pour mettre à jour votre index de paquets et installer ‘gnupg’ et ‘curl’ sur votre système.

sudo apt update && sudo apt install gnupg curl -y

Exécutez la commande ci-dessous pour ajouter la clé GPG pour le serveur MongoDB.

curl -fsSL https://www.mongodb.org/static/pgp/server-8.0.asc | \  
sudo gpg -o /usr/share/keyrings/mongodb-server-8.0.gpg \  
--dearmor

Ajoutez le dépôt MongoDB en utilisant la commande ci-dessous.

echo "deb [ arch=amd64,arm64 signed-by=/usr/share/keyrings/mongodb-server-8.0.gpg ] https://repo.mongodb.org/apt/ubuntu noble/mongodb-org/8.0 multiverse" | sudo tee /etc/apt/sources.list.d/mongodb-org-8.0.list

Après l’ajout du dépôt, exécutez la commande suivante pour rafraîchir votre index de paquets et installer le serveur MongoDB. Entrez ‘Y’ pour confirmer l’installation.

sudo apt update && sudo apt install mongodb-org

Lorsque l’installation est terminée, démarrez et activez le service MongoDB ‘mongod’, puis vérifiez l’état de votre service MongoDB pour vous assurer qu’il fonctionne.

sudo systemctl enable --now mongod  
sudo systemctl status mongod

Vous pouvez voir ci-dessous que le serveur MongoDB fonctionne.

De plus, vous pouvez vous connecter au serveur MongoDB avec la commande ‘mongosh’ ci-dessous. Pour quitter, appuyez sur Ctrl+d.

mongosh

Configuration de Python et de l’environnement virtuel

Avec MongoDB installé, vous allez installer des paquets Python et configurer le répertoire du projet et l’environnement virtuel.

Installez Python, Pip et les modules Venv avec la commande suivante. Entrez ‘Y’ pour confirmer l’installation.

sudo apt install python3 python3-pip python3-venv

Une fois l’installation terminée, connectez-vous à votre utilisateur.

su - username

Maintenant, créez un nouveau répertoire ‘~/app‘ et déplacez-vous dedans. Ce répertoire sera utilisé pour stocker votre projet FastAPI.

mkdir -p ~/app; cd ~/app

Exécutez la commande ci-dessous pour créer un nouvel environnement virtuel ‘venv‘ et l’activer. Avec cela, votre invite de commande deviendra comme ‘(venv) user@hostname‘.

python3 -m venv .venv  
source .venv/bin/activate

À partir de là, votre environnement de travail doit être sur l’environnement virtuel ‘venv’. Vous pouvez vous déconnecter de ‘venv’ en utilisant la commande ci-dessous.

deactivate

Création du projet FastAPI

Maintenant que vous avez créé et activé votre environnement virtuel Python, installons FastAPI et créons la structure du projet.

Avec la commande ‘pip3’, exécutez-la pour installer les paquets ‘fastapi’ et ‘uvicorn’.

pip3 install fastapi uvicorn

• Le ‘fastapi’ est le principal framework web FastAPI pour construire des API en Python
• Le ‘uvicorn’ est l’implémentation du serveur web ASGI (Asynchronous Server Gateway Interface) en Python.

Après la fin de l’installation, créez de nouveaux fichiers et répertoires avec la commande suivante.

mkdir -p server/{models,routes}  
touch main.py server/{app.py,database.py} server/models/itemModels.py server/routes/item.py

Voici la structure de notre projet FastAPI.

server/app.py

Maintenant que votre projet est prêt, modifions le fichier ‘server/app.py’, qui est l’application principale de votre projet FastAPI.

Ouvrez le fichier ‘app.py‘ avec votre éditeur de texte et copiez le script suivant.

from fastapi import FastAPI  
  
app = FastAPI()  
  
@app.get("/")  
async def root():  
 return {"message": "Hello FastAPI!"}

• Importez le module FastAPI dans votre projet et liez-le à la variable ‘app’
• Créez une nouvelle fonction ‘root’ qui renvoie ‘Hello FastAPI!’
• La fonction ‘root’ répond à la méthode GET sur l’URL racine
• ‘async’ marque votre fonction comme une fonction asynchrone et peut utiliser ‘await’ dans son corps lorsqu’elle est appelée

main.py

Dans cette section, nous allons modifier le fichier ‘main.py’ qui sera utilisé pour exécuter votre projet FastAPI via ‘uvicorn’ (serveur web ASGI en Python).

Maintenant, ouvrez et modifiez le script ‘main.py‘ et insérez le code suivant.

import uvicorn  
  
if __name__ == "__main__":  
 uvicorn.run("server.app:app", host="0.0.0.0", port=8080, reload=True)

• Importez le module ‘uvicorn’
• Lorsque le script ‘main.py’ est exécuté, il chargera le module ‘app’ ou FastAPI dans ‘server/app.py’
• FastAPI sera exécuté sur ‘0.0.0.0’ avec le port ‘8080’
• Activez le rechargement automatique lorsque le code change via ‘reload=True’

Exécutez votre projet FastAPI

Maintenant que votre projet est prêt, exécutons votre premier projet FastAPI.

Exécutez le script ‘main.py’ comme suit et votre FastAPI sera en cours d’exécution sur votre système.

python3 main.py

Maintenant, ouvrez votre navigateur web et visitez http://SERVERIP:8080/. Si votre installation est réussie, vous verrez le message ‘Hello FastAPI!’. Vous pouvez également y accéder via ‘curl’ depuis le terminal.

Enfin, vous pouvez accéder à la documentation de l’API sur http://SERVERIP:8080/docs > qui est fournie par Swagger UI.

Connexion de FastAPI à MongoDB

Dans ce guide, vous allez créer une API de base avec FastAPI et MongoDB. Votre API devrait être capable de CRUD avec le serveur de base de données MongoDB. Pour cette étape, vous allez connecter votre projet au serveur MongoDB.

Tout d’abord, exécutez la commande ‘pip3’ ci-dessous pour installer le pilote MongoDB ‘motor‘ dans votre projet. ‘motor’ fournit un accès API non-bloquant et basé sur des coroutines au serveur MongoDB.

pip3 install motor

server/database.py

Après l’installation du module ‘motor‘, modifions le script ‘server/database.py‘.

Ouvrez le fichier ‘server/database.py’ avec votre éditeur de texte et saisissez le script suivant. Cela sera utilisé pour se connecter au serveur MongoDB via le module ‘motor’.

from motor.motor_asyncio import AsyncIOMotorClient  
  
MONGODB_HOST = "mongodb://localhost:27017"  
  
connection = AsyncIOMotorClient(MONGODB_HOST)  
  
database = connection.items  
item_collection = database.get_collection("item_collection")

• Importez ‘AsyncIOMotorClient’ depuis ‘motor.motor_asyncio’
• Créez une nouvelle constante ‘MONGODB_HOST’ et pointez vers le serveur MongoDB ‘mongodb://localhost:27017’
• Connectez-vous au serveur MongoDB via la variable ‘connection’
• Connectez-vous à la base de données ‘items’ via la variable ‘database’
• Accédez aux collections de la base de données avec la variable ‘item_collection’

Création du modèle de base de données avec pydantic

Dans cette section, vous allez concevoir vos données via ‘pydantic‘, qui fournit un modélisation pour notre base de données MongoDB.

Installez le module ‘pydantic’ avec la commande ‘pip3’ ci-dessous. Le module ‘pydantic‘ est une bibliothèque de validation de données qui vous permet de créer un schéma de base de données via un modèle.

pip3 install pydantic

Maintenant, ouvrez le fichier ‘server/models/itemModels.py‘ avec votre éditeur de texte et copiez le script suivant.

from pydantic import BaseModel, Field  
from typing import Optional  
  
class Item(BaseModel):  
 name: str  
 category: str  
 stocks: int  
 price: int = Field(gt=0)  
  
 class Config:  
 json_schema_extra = {  
 "example": {  
 "name": "Company Smart Watch",  
 "category": "smartwatch",  
 "stocks": 10,  
 "price": 1000,  
 }  
 }  
  
class ItemUpdate(BaseModel):  
 name: Optional[str] = None  
 category: Optional[str] = None  
 stocks: Optional[int] = None  
 price: Optional[int] = None  
  
 class Config:  
 json_schema_extra = {  
 "example": {  
 "name": "New Smart watch",  
 "category": "new-smartwatch",  
 "stocks": 5,  
 "price": 500,  
 }  
 }

• Importez ‘BaseModel’ et ‘Field’ depuis ‘pydantic’
• Importez le module ‘Optional’ depuis ‘typing’
• Créez le schéma de base de données ‘Item’ suivant pour FastAPI : - ‘name’ et ‘category’ avec le type chaîne
• ‘stocks’ et ‘price’ avec le type entier
• ‘price’ doit être supérieur à 0
• Étendez le modèle de données via la classe ‘Config’ en fournissant un exemple de données que l’utilisateur peut inclure dans la requête
• Créez le schéma ‘ItemUpdate’ suivant avec chaque champ comme optionnel.

Ajout des opérations CRUD

À ce stade, vous avez créé une connexion au serveur MongoDB et créé un schéma de base de données via ‘pydantic’. Passons maintenant à la création des opérations CRUD avec FastAPI et MongoDB.

Ouvrez à nouveau le script ‘server/database.py‘ et créez une nouvelle fonction ‘item_helper‘ de type ‘dict‘. Appelez cette fonction pour obtenir des données détaillées sur un article.

def item_helper(item) -> dict:  
 return {  
 "id": str(item["_id"]),  
 "name": item["name"],  
 "category": item["category"],  
 "stocks": item["stocks"],  
 "price": item["price"],  
 }

Ajoutez la ligne suivante pour importer le module ‘ObjectId‘ depuis ‘bson.objectid’. Le ‘ObjectId‘ fait partie du type de données dans BSON, que MongoDB utilise pour le stockage de données et la représentation des données JSON.

from bson.objectid import ObjectId

Créer un article

Tout d’abord, vous allez créer une fonction asynchrone qui sera capable d’ajouter de nouvelles données à la base de données MongoDB.

Créez une nouvelle fonction ‘add_item‘ comme suit :

# Ajouter un nouvel article  
async def add_item(item_details: dict) -> dict :  
 item = await item_collection.insert_one(item_details)  
 new_item = await item_collection.find_one({"_id": item.inserted_id})  
 return item_helper(new_item)

• La fonction ‘add_item’ accepte les données sous forme de dictionnaire de vos détails d’article
• Via ‘item_collection’ pour accéder aux documents dans la base de données et exécuter la requête ‘insert_one’ pour ajouter un nouvel article
• Le nouvel article sera affiché après une opération réussie

Récupérer tous les articles

Deuxièmement, vous allez créer une nouvelle fonction ‘get_items’ qui sera capable de récupérer tous les articles disponibles dans votre base de données.

Créez une nouvelle fonction ‘get_items‘ pour récupérer tous les articles de MongoDB.

# récupérer tous les articles  
async def get_items():  
 items = []  
 async for item in item_collection.find():  
 items.append(item_helper(item))  
 return items

• Vous allez créer une liste vide ‘items’
• En utilisant la requête ‘find()’ pour rechercher toutes les données et boucler à travers elles
• Chaque article sera ajouté à la liste ‘items’ via la méthode ‘append’
• Une fois la boucle terminée, vos articles seront affichés

Récupérer un article spécifique basé sur l’id

Ensuite, vous allez créer une nouvelle fonction qui sera capable de récupérer des données pour un sélecteur spécifique ‘id’. Cela vous montrera des données détaillées sur des articles spécifiques.

Créez une nouvelle fonction ‘get_item‘ pour récupérer les données d’articles spécifiques. La fonction ‘get_item’ acceptera une chaîne ‘id’ comme sélecteur et renverra un dictionnaire.

# récupérer un article spécifique  
async def get_item(id: str) -> dict:  
 item = await item_collection.find_one({"_id": ObjectId(id)})  
 if item:  
 return item_helper(item)  
 return "Article non trouvé."

• La requête ‘find_one()’ sera utilisée pour récupérer des données basées sur l’id
• Si un article est trouvé, les détails seront affichés en utilisant le format de dictionnaire ‘item_helper’
• Si l’article n’est pas disponible, la réponse est ‘Article non trouvé’

Mettre à jour un article

Maintenant, vous allez créer une nouvelle fonction pour mettre à jour un article. Vous pouvez également mettre à jour partiellement votre article via l’API.

Définissez la nouvelle fonction ‘change_item‘ comme suit :

# mettre à jour l'article  
async def change_item(id: str, data: dict):  
 if len(data) < 1:  
 return "Veuillez saisir vos données"  
 find_item = await item_collection.find_one({"_id": ObjectId(id)})  
 if find_item:  
 item_update = await item_collection.update_one({"_id": ObjectId(id)}, {"$set": data})  
 if item_update:  
 return True  
 return False

• La fonction ‘change_item’ prend deux arguments, l’id de l’article cible et les données comme nouvelles données
• Si les données sont vides ou ‘< 1’, l’opération est terminée
• Cette fonction trouvera un article basé sur le sélecteur ‘id’
• Si l’id est trouvé, ‘item_update’ sera exécuté
• Si ‘item_update’ réussit, renvoyez ‘True’, ou renvoyez ‘False’ sinon

Supprimer un article

Enfin, vous allez créer la fonction ‘delete_item’ pour supprimer des articles via un sélecteur spécifique.

Pour supprimer un article, nous allons créer la fonction ‘delete_item‘ et utiliser le sélecteur ‘id’ comme suit :

# supprimer un article  
async def delete_item(id: str):  
 item = await item_collection.find_one({"_id": ObjectId(id)})  
 if item:  
 await item_collection.delete_one({"_id": ObjectId(id)})  
 return(f'Article {id} supprimé.')  
 return "Article non trouvé."

• La requête ‘find_one()’ recherchera un article basé sur l’id fourni
• Si un article est trouvé, il sera supprimé via la requête ‘delete_one()’ et renverra ‘Article id supprimé’
• Si l’article n’est pas disponible, ‘Article non trouvé’ sera affiché

Ajout de routes pour les opérations CRUD

À ce stade, nous avons créé des fonctions asynchrones pour l’opération CRUD avec FastAPI. Maintenant, créons la route ou le point de terminaison pour chaque opération.

Modifiez le fichier ‘server/routes/item.py‘ en utilisant votre éditeur préféré.

server/routes/item.py

Tout d’abord, ajoutez les modules ‘APIRouter‘ et ‘Body‘ depuis ‘fastapi’. Ensuite, ajoutez le ‘jsonable_encode‘ depuis ‘fastapi.encoders’.

from fastapi import APIRouter, Body  
from fastapi.encoders import jsonable_encoder

Importez chaque fonction depuis le fichier ‘database.py’.

from server.database import (  
 add_item,  
 get_items,  
 get_item,  
 change_item,  
 delete_item,  
)

Importez les modèles ‘Item’ et ‘ItemUpdate’ depuis le fichier ‘itemModels.py’.

from server.models.itemModels import (  
 Item,  
 ItemUpdate,  
)

Appelez la classe ‘APIRouter’ via la variable ‘router’.

router = APIRouter()

Route pour ajouter un nouvel article

Maintenant, ajoutons une route pour ajouter de nouveaux articles. Dans cet exemple, vous pouvez ajouter un nouvel article via POST à l’URL ‘/item‘.

Ajoutez les lignes suivantes pour configurer les routes pour ajouter de nouveaux articles. Chaque POST à l’URL racine de l’article sera traité comme l’insertion de nouvelles données.

# opération d'ajout d'un nouvel article  
@router.post("/")  
async def add_item_data(item: Item = Body(...)):  
 item = jsonable_encoder(item)  
 new_item = await add_item(item)  
 return new_item

• La fonction ‘add_item_data’ accepte le schéma ‘Item’, qui fera partie du ‘Body’ de votre requête
• Votre article sera converti en format dictionnaire via ‘jsonable_encoder’
• Insérez vos données sous forme de dictionnaire via la fonction ‘add_item’ (voir database.py) sur la variable ‘new_item’
• Retournez vos données insérées ‘new_item’ en tant que réponse

Route pour obtenir tous les articles

Ajoutez le script suivant pour configurer une route pour récupérer des données. Chaque requête GET à l’URL racine de l’article récupérera toutes vos données.

# obtenir tous les articles disponibles  
@router.get("/")  
async def get_item_data():  
 items = await get_items()  
 if items:  
 return items  
 return "Aucun article disponible."

• Créez la fonction ‘get_item_data’ qui exécutera la fonction ‘get_item’ depuis le fichier ‘database.py’
• Si des articles sont disponibles, vous obtiendrez la liste de tous vos articles
• Si aucun article, vous obtiendrez la réponse ‘Aucun article disponible’

Route pour obtenir un article spécifique

Pour obtenir les détails d’un article spécifique, nous utiliserons l’id comme sélecteur. Chaque requête GET à ‘/id’ renverra un article détaillé de l’id demandé.

# Afficher les détails de l'article via l'id  
@router.get("/{id}")  
async def get_item_details(id):  
 item_details = await get_item(id)  
 if item_details:  
 return item_details  
 return "Article non trouvé."

• La fonction ‘get_item_details’ sera créée et passera l’id depuis l’URL
• La fonction ‘get_item’ (voir database.py) sera appelée et passera également l’id comme argument
• Si l’article est trouvé, les détails de l’article seront affichés
• Si aucun article avec cet id spécifique, vous obtiendrez ‘Article non trouvé’

Route pour mettre à jour un article

Copiez le code suivant pour définir la route pour l’article mis à jour :

# Mettre à jour l'article  
@router.put("/{id}")  
async def update_item(id: str, data: ItemUpdate = Body(...)):  
 data = {k: v for k, v in data.dict().items() if v is not None}  
 updated_item = await change_item(id, data)  
 if updated_item:  
 return{f'Success: article {id} mis à jour.'}  
 return "Erreur"

• La fonction ‘update_item’ prendra deux arguments, ‘id’ comme sélecteur et les ‘data’, qui sont basées sur le modèle ‘ItemUpdate’
• Les données seront vérifiées dans la variable ‘data’
• L’ ‘updated_item’ exécutera la fonction ‘change_item’ depuis le fichier ‘database.py’
• Si la mise à jour réussit, vous verrez la sortie ‘Succès’

Route pour supprimer un article

Insérez les lignes suivantes pour créer la fonction ‘remove_item’ pour supprimer des articles. Chaque opération DELETE à un ‘/id’ spécifique supprimera l’article correspondant à l’id.

# supprimer l'article via l'id  
@router.delete("/{id}")  
async def remove_item(id):  
 item_to_delete = await delete_item(id)  
 if item_to_delete:  
 return item_to_delete  
 return{f'Article {id} non disponible.'}

• La fonction ‘remove_item’ exécutera ‘delete_item’ et passera le sélecteur ‘id’
• L’opération de suppression sera stockée et exécutée via la variable ‘item_to_delete’
• Lorsque l’article n’est pas disponible, vous obtiendrez le retour ‘Article id non disponible’

server/app.py

Maintenant que vous avez terminé le fichier ‘server/routes/item.py’, incluons-le dans le ‘server/app.py’.

Ouvrez le fichier ‘app.py‘ avec votre éditeur de texte.

Importez le ‘router’ depuis le fichier ‘server/routes/item.py‘ en tant que ‘ItemRouter’.

from server.routes.item import router as ItemRouter

Insérez le ‘ItemRouter’ avec le préfixe par défaut ‘/item‘. Les opérations CRUD seront gérées via l’URL ‘/item‘.

app.include_router(ItemRouter, tags=["Item"], prefix="/item")

Maintenant, votre point de terminaison CRUD sera disponible à l’adresse suivante :

• Ajouter un nouvel article : POST à ‘/item‘
• Récupérer tous les articles : GET à ‘/item‘
• Récupérer un article spécifique : GET à ‘/item/id’. L’id est généré par MongoDB
• Mettre à jour un article : PUT à ‘/item/id’
• Supprimer un article : DELETE à ‘/item/id’

Tester les opérations CRUD

Tout d’abord, assurez-vous que votre projet FastAPI est en cours d’exécution, ou vous pouvez exécuter le script ‘main.py’ comme suit :

python3 main.py

Naviguez à travers http://SERVERIP:8080/docs et vous verrez chaque route que vous avez créée.

Voici un exemple d’ajout d’un nouvel article.

Récupérez tous les articles disponibles via l’API.

Récupérez des articles spécifiques via le sélecteur ‘id’.

Mettez à jour des données partielles d’articles spécifiques.

Voici les données mises à jour.

Voici l’opération de suppression via le sélecteur ‘id’.

Conclusion

Félicitations ! Vous avez terminé l’installation de FastAPI avec MongoDB sur Ubuntu 24.04. Vous avez également appris comment connecter FastAPI à MongoDB en utilisant le module ‘motor’, créer des modèles de données via ‘pydantic’, créer des opérations CRUD avec FastAPI, et apprendre à créer des points de terminaison pour votre API.