Microservices avec FastAPI
Pourquoi Microservices avec FastAPI ?
Dans un monde d'architecture monolithique, chaque fonctionnalité est souvent développée et déployée en une seule fois. Cependant, cette approche présente plusieurs problèmes : l'intégration difficile des nouvelles fonctionnalités, les longues délais de déploiement et le risque élevé de pannes système.
Un microservice est une unité autonome qui répond à un besoin spécifique et peut être développée, déployée et évoluée indépendamment d'autres services. En utilisant FastAPI pour développer des microservices, on obtient un environnement plus modulaire et flexible, ce qui facilite la maintenance, l'évolution et le déploiement.
Un cas concret est une application e-commerce. Chaque composante de l'application (front-end, backend, gestion des paiements, stockage) peut être développée en tant que microservice indépendant, ce qui permet un développement plus rapide et la mise à jour individuelle des fonctionnalités.
Prerequis
Pour suivre ce tutoriel, vous devrez avoir les connaissances suivantes :
- Connaissance avancée de Python
- Familiarité avec le concept de microservices
- Connaissance de FastAPI pour le développement web asynchrone en Python
- Installation d'un environnement Python (Python 3.7+ recommandé)
- Un éditeur de code (VSCode, PyCharm, etc.)
- Docker pour l'orchestration des services
Installez les dépendances nécessaires :
pip install fastapi uvicorn sqlalchemy databases asyncpg alembic docker-compose
Concepts fondamentaux
1. Un microservice est une unité autonome qui répond à un besoin spécifique.
Un microservice doit être capable de fonctionner indépendamment des autres services, avec sa propre base de données et son propre ensemble de ressources.
## app.py pour notre microservice "task_manager"
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"message": "Task Manager API"}
2. Chaque microservice doit être capable de communiquer avec d'autres services via des interfaces standardisées.
Pour ce faire, on utilise principalement les APIs REST ou gRPC.
## app.py pour notre microservice "user_manager"
from fastapi import FastAPI
import requests
app = FastAPI()
@app.get("/tasks/{task_id}")
async def get_task(task_id: int):
user_response = requests.get(f"http://user_manager/users/{task_id}/owner")
task_owner = user_response.json()
return {"task_id": task_id, "owner": task_owner}
3. Utilisez un orchestrateur pour gérer les services et leur déploiement.
Docker-compose est une excellent solution pour ce cas. Voici un exemple de docker-compose.yml :
version: '3.8'
services:
user_manager:
build: ./user_manager
ports:
- "8001:8000"
task_manager:
build: ./task_manager
ports:
- "8002:8000"
4. Gestion des bases de données pour chaque microservice.
Chaque service doit avoir sa propre base de données, ce qui permet une gestion plus fine et évite les conflits.
## models.py pour notre microservice "task_manager"
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@db:5432/task_manager"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True, index=True)
description = Column(String, index=True)
Mise en pratique : projet fil rouge
Nous allons créer un gestionnaire de tâches simple avec deux microservices : user_manager et task_manager.
Étape 1 : Création du service user_manager
Créez un répertoire pour le service user_manager :
mkdir user_manager
cd user_manager
Créer les fichiers suivants :
app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
users = {}
@app.post("/users/", status_code=201)
async def create_user(user: BaseModel):
if user.id in users:
raise HTTPException(status_code=400, detail="User already exists")
users[user.id] = {"id": user.id, "name": user.name}
return users[user.id]
@app.get("/users/{user_id}")
async def read_user(user_id: int):
if user_id not in users:
raise HTTPException(status_code=404, detail="User not found")
return users[user_id]
models.py
class User(BaseModel):
id: int
name: str
Dockerfile
FROM python:3.8-slim
WORKDIR /app
COPY . /app
RUN pip install fastapi uvicorn pydantic
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]
Étape 2 : Création du service task_manager
Créez un répertoire pour le service task_manager :
mkdir task_manager
cd task_manager
Créer les fichiers suivants :
app.py
from fastapi import FastAPI, HTTPException
import requests
app = FastAPI()
@app.post("/tasks/", status_code=201)
async def create_task(task: dict):
user_response = requests.get(f"http://user_manager/users/{task['owner']}")
if user_response.status_code == 404:
raise HTTPException(status_code=400, detail="User does not exist")
return {"id": task["id"], "description": task["description"], "owner": task["owner"]}
@app.get("/tasks/{task_id}")
async def read_task(task_id: int):
user_response = requests.get(f"http://user_manager/users/{task_id}/owner")
if user_response.status_code == 404:
raise HTTPException(status_code=404, detail="Task not found")
return {"id": task_id, "description": "Sample task", "owner": "1"}
Dockerfile
FROM python:3.8-slim
WORKDIR /app
COPY . /app
RUN pip install fastapi uvicorn requests
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]
Étape 3 : Configuration de docker-compose.yml
Créez un fichier docker-compose.yml dans le répertoire racine :
version: '3.8'
services:
user_manager:
build: ./user_manager
ports:
- "8001:8000"
task_manager:
build: ./task_manager
ports:
- "8002:8000"
Étape 4 : Lancement des services
Exécutez les commandes suivantes pour démarrer les services :
docker-compose up --build
Vous devriez maintenant avoir deux microservices en cours d'exécution sur vos ports respectifs.
Erreurs frequentes et debugging
1. requests.exceptions.ConnectionError
Code incorrect :
import requests
response = requests.get("http://user_manager/users/1")
Code correct :
import requests
response = requests.get("http://user_manager:8000/users/1")
2. KeyError dans les dictionnaires
Code incorrect :
users = {}
user_id = 1
print(users[user_id])
Code correct :
users = {}
user_id = 1
if user_id in users:
print(users[user_id])
else:
print("User not found")
3. HTTPException non capturée
Code incorrect :
from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.get("/users/{user_id}")
async def read_user(user_id: int):
if user_id not in users:
raise HTTPException(status_code=404, detail="User not found")
return users[user_id]
Code correct :
from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.get("/users/{user_id}")
async def read_user(user_id: int):
if user_id not in users:
raise HTTPException(status_code=404, detail="User not found")
return {"id": user_id, "name": users[user_id]}
Pour aller plus loin
1. Utilisation d'un système de service discovery pour gérer les adresses des services.
Docker Compose et Docker Swarm sont capables de gérer les adresses IP des services, mais pour une architecture plus large, on peut utiliser un système comme Consul ou Eureka.
2. Sécurité et authentification
Pour assurer la sécurité de vos microservices, vous pouvez utiliser des solutions comme OAuth2, JWT ou Keycloak.
3. Métriques et logs
La supervision des métriques et des logs est essentielle pour le maintien de l'application. Utilisez Prometheus, Grafana, ELK Stack, etc.
Défi pratique
Implémentez un système d'authentification JWT pour les microservices user_manager et task_manager. Créez une route /login dans user_manager qui retourne un token JWT et utilisez ce token pour sécuriser les routes des deux services.
N'oubliez pas de tester votre application après chaque modification pour vous assurer qu'elle fonctionne correctement.