BONUS : FastAPI pour Data Engineers

Ce module bonus couvre FastAPI, le framework Python moderne pour créer des APIs REST performantes. En tant que Data Engineer, tu auras souvent besoin d'exposer des données via des APIs.

Prérequis

Objectifs du module

À la fin de ce notebook, tu seras capable de :

• Créer une API REST avec FastAPI
• Définir des modèles de données avec Pydantic
• Créer des endpoints CRUD (Create, Read, Update, Delete)
• Valider automatiquement les données entrantes
• Servir des données Pandas via API
• Connecter une API à une base de données
• Générer une documentation Swagger automatique
• Déployer une API avec Uvicorn

💡 Note : FastAPI s'exécute comme un serveur. Certains exemples seront à tester hors du notebook, dans des fichiers .py.

Pourquoi FastAPI pour un Data Engineer ?

┌─────────────────────────────────────────────────────────────────┐
│              CAS D'USAGE DATA ENGINEERING                       │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│     Exposer des données traitées (datamart → API)               │
│     Créer des webhooks pour déclencher des pipelines            │
│     Servir des prédictions ML en production                     │
│     Fournir des métriques pour les dashboards                   │
│     Créer des microservices data                                │
│     Valider des données avant ingestion                         │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

FastAPI vs autres frameworks

💡 FastAPI combine le meilleur : performance, simplicité, et documentation automatique.

Installation

# Installation de base
pip install fastapi uvicorn

# Avec toutes les dépendances optionnelles
pip install "fastapi[all]"

# Installation
!pip install fastapi uvicorn pydantic pandas --quiet

print("Packages installés !")

1️⃣ Ta première API FastAPI

Structure minimale

Créer un fichier main.py :

# main.py
from fastapi import FastAPI

# Créer l'application
app = FastAPI()

# Premier endpoint
@app.get("/")
def root():
    return {"message": "Hello Data Engineer!"}

@app.get("/health")
def health_check():
    return {"status": "healthy"}

Lancer le serveur

# Dans le terminal
uvicorn main:app --reload

# main = fichier main.py
# app = variable FastAPI()
# --reload = redémarre automatiquement si le code change

Résultat :

INFO:     Uvicorn running on http://127.0.0.1:8000
INFO:     Started reloader process

Tester l'API

Les méthodes HTTP

┌─────────────────────────────────────────────────────────────────┐
│                    MÉTHODES HTTP (CRUD)                         │
├──────────┬──────────────────┬───────────────────────────────────┤
│ Méthode  │ Action           │ Exemple                           │
├──────────┼──────────────────┼───────────────────────────────────┤
│ GET      │ Lire (Read)      │ GET /users → Liste des users      │
│ POST     │ Créer (Create)   │ POST /users → Créer un user       │
│ PUT      │ Remplacer        │ PUT /users/1 → Remplacer user 1   │
│ PATCH    │ Modifier         │ PATCH /users/1 → Modifier user 1  │
│ DELETE   │ Supprimer        │ DELETE /users/1 → Supprimer       │
└──────────┴──────────────────┴───────────────────────────────────┘

from fastapi import FastAPI

app = FastAPI()

@app.get("/items")
def get_items():
    return {"action": "Liste des items"}

@app.post("/items")
def create_item():
    return {"action": "Création d'un item"}

@app.put("/items/{item_id}")
def update_item(item_id: int):
    return {"action": f"Mise à jour de l'item {item_id}"}

@app.delete("/items/{item_id}")
def delete_item(item_id: int):
    return {"action": f"Suppression de l'item {item_id}"}

Paramètres de chemin (Path Parameters)

Les paramètres dans l'URL permettent d'identifier une ressource spécifique.

from fastapi import FastAPI

app = FastAPI()

# Paramètre simple
@app.get("/users/{user_id}")
def get_user(user_id: int):  # Typage automatique !
    return {"user_id": user_id}

# Plusieurs paramètres
@app.get("/users/{user_id}/orders/{order_id}")
def get_user_order(user_id: int, order_id: int):
    return {
        "user_id": user_id,
        "order_id": order_id
    }

Validation automatique

GET /users/42        → {"user_id": 42}       ✅
GET /users/abc       → Erreur 422            ❌ (pas un int)
GET /users/-1        → {"user_id": -1}       ✅ (mais logiquement faux)

Validation avancée avec Path

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/users/{user_id}")
def get_user(
    user_id: int = Path(
        ...,  # Requis
        title="User ID",
        description="L'identifiant unique de l'utilisateur",
        ge=1,  # Greater or Equal (>= 1)
        le=10000  # Less or Equal (<= 10000)
    )
):
    return {"user_id": user_id}

Paramètres de requête (Query Parameters)

Les paramètres après le ? dans l'URL pour filtrer, paginer, etc.

GET /users?skip=0&limit=10&active=true
         └──────────┬──────────────┘
              Query parameters

from fastapi import FastAPI
from typing import Optional

app = FastAPI()

@app.get("/users")
def get_users(
    skip: int = 0,           # Défaut = 0
    limit: int = 10,         # Défaut = 10
    active: bool = True,     # Défaut = True
    search: Optional[str] = None  # Optionnel
):
    return {
        "skip": skip,
        "limit": limit,
        "active": active,
        "search": search
    }

Cas d'usage Data Engineering : Pagination

@app.get("/data")
def get_data(
    page: int = 1,
    page_size: int = 100,
    sort_by: str = "created_at",
    order: str = "desc"
):
    # Calculer l'offset
    offset = (page - 1) * page_size
    
    return {
        "page": page,
        "page_size": page_size,
        "offset": offset,
        "sort_by": sort_by,
        "order": order
    }

2️⃣ Pydantic — Validation des données

Pydantic est au cœur de FastAPI. Il permet de :

• Définir des schémas de données
• Valider automatiquement les entrées
• Convertir les types
• Générer la documentation

Définir un modèle

from pydantic import BaseModel, Field, EmailStr
from typing import Optional
from datetime import datetime

class User(BaseModel):
    """Modèle utilisateur pour l'API."""
    
    id: Optional[int] = None
    nom: str = Field(..., min_length=2, max_length=50)
    email: EmailStr
    age: int = Field(..., ge=0, le=150)
    actif: bool = True
    created_at: datetime = Field(default_factory=datetime.now)
    
    class Config:
        # Exemple pour la documentation
        json_schema_extra = {
            "example": {
                "nom": "Alice Dupont",
                "email": "alice@example.com",
                "age": 30,
                "actif": True
            }
        }

Validateurs disponibles

# Exemple Pydantic dans le notebook
from pydantic import BaseModel, Field, field_validator
from typing import Optional
from datetime import datetime

class Transaction(BaseModel):
    """Modèle de transaction pour pipeline ETL."""
    
    id: Optional[int] = None
    montant: float = Field(..., gt=0, description="Montant positif")
    devise: str = Field(..., min_length=3, max_length=3)
    description: str = Field(..., min_length=1, max_length=200)
    timestamp: datetime = Field(default_factory=datetime.now)
    
    @field_validator('devise')
    @classmethod
    def devise_uppercase(cls, v):
        return v.upper()

# Test de validation
try:
    # Valide
    tx1 = Transaction(montant=100.50, devise="eur", description="Achat")
    print(f" Transaction valide : {tx1}")
    print(f" Devise convertie : {tx1.devise}")
    
    # Invalide
    tx2 = Transaction(montant=-50, devise="EUR", description="Test")
except Exception as e:
    print(f"\n Erreur de validation : {e}")

Corps de requête (Request Body)

Pour les requêtes POST/PUT, les données sont envoyées dans le body (corps) en JSON.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserCreate(BaseModel):
    nom: str
    email: str
    age: int

class UserResponse(BaseModel):
    id: int
    nom: str
    email: str
    age: int

# Base de données fictive
users_db = []
user_id_counter = 1

@app.post("/users", response_model=UserResponse)
def create_user(user: UserCreate):
    global user_id_counter
    
    # Créer le user avec un ID
    new_user = {
        "id": user_id_counter,
        "nom": user.nom,
        "email": user.email,
        "age": user.age
    }
    
    users_db.append(new_user)
    user_id_counter += 1
    
    return new_user

Requête avec curl

curl -X POST "http://127.0.0.1:8000/users" \
  -H "Content-Type: application/json" \
  -d '{"nom": "Alice", "email": "alice@test.com", "age": 30}'

Requête avec Python requests

import requests

response = requests.post(
    "http://127.0.0.1:8000/users",
    json={"nom": "Alice", "email": "alice@test.com", "age": 30}
)
print(response.json())

3️⃣ API CRUD complète

Voici une API complète avec toutes les opérations CRUD :

# crud_api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

app = FastAPI(
    title="Data Pipeline API",
    description="API pour gérer les pipelines de données",
    version="1.0.0"
)

# ============ MODÈLES ============

class PipelineBase(BaseModel):
    nom: str = Field(..., min_length=3, max_length=100)
    description: Optional[str] = None
    schedule: str = Field(..., description="Cron expression")
    actif: bool = True

class PipelineCreate(PipelineBase):
    pass

class PipelineUpdate(BaseModel):
    nom: Optional[str] = None
    description: Optional[str] = None
    schedule: Optional[str] = None
    actif: Optional[bool] = None

class Pipeline(PipelineBase):
    id: int
    created_at: datetime
    updated_at: datetime

# ============ BASE DE DONNÉES (simulée) ============

pipelines_db: dict[int, dict] = {}
pipeline_id_counter = 1

# ============ ENDPOINTS ============

# CREATE
@app.post("/pipelines", response_model=Pipeline, status_code=201)
def create_pipeline(pipeline: PipelineCreate):
    global pipeline_id_counter
    
    now = datetime.now()
    new_pipeline = {
        "id": pipeline_id_counter,
        **pipeline.model_dump(),
        "created_at": now,
        "updated_at": now
    }
    
    pipelines_db[pipeline_id_counter] = new_pipeline
    pipeline_id_counter += 1
    
    return new_pipeline

# READ ALL
@app.get("/pipelines", response_model=list[Pipeline])
def list_pipelines(skip: int = 0, limit: int = 10, actif: Optional[bool] = None):
    pipelines = list(pipelines_db.values())
    
    # Filtrer par statut actif
    if actif is not None:
        pipelines = [p for p in pipelines if p["actif"] == actif]
    
    return pipelines[skip:skip + limit]

# READ ONE
@app.get("/pipelines/{pipeline_id}", response_model=Pipeline)
def get_pipeline(pipeline_id: int):
    if pipeline_id not in pipelines_db:
        raise HTTPException(status_code=404, detail="Pipeline non trouvé")
    return pipelines_db[pipeline_id]

# UPDATE
@app.patch("/pipelines/{pipeline_id}", response_model=Pipeline)
def update_pipeline(pipeline_id: int, pipeline: PipelineUpdate):
    if pipeline_id not in pipelines_db:
        raise HTTPException(status_code=404, detail="Pipeline non trouvé")
    
    stored = pipelines_db[pipeline_id]
    update_data = pipeline.model_dump(exclude_unset=True)
    
    for key, value in update_data.items():
        stored[key] = value
    
    stored["updated_at"] = datetime.now()
    
    return stored

# DELETE
@app.delete("/pipelines/{pipeline_id}", status_code=204)
def delete_pipeline(pipeline_id: int):
    if pipeline_id not in pipelines_db:
        raise HTTPException(status_code=404, detail="Pipeline non trouvé")
    
    del pipelines_db[pipeline_id]
    return None

Gestion des erreurs

FastAPI utilise HTTPException pour retourner des erreurs HTTP propres.

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

@app.get("/users/{user_id}")
def get_user(user_id: int):
    user = find_user_by_id(user_id)  # Fonction fictive
    
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"User {user_id} non trouvé",
            headers={"X-Error": "User not found"}
        )
    
    return user

Codes HTTP courants

4️⃣ Servir des données Pandas via API

C'est le cas d'usage le plus courant pour un Data Engineer : exposer des données traitées.

# data_api.py
from fastapi import FastAPI, Query
from fastapi.responses import JSONResponse, StreamingResponse
import pandas as pd
import io

app = FastAPI(title="Data API")

# Charger les données (en production: depuis une DB ou un fichier)
df = pd.DataFrame({
    "id": range(1, 101),
    "nom": [f"User_{i}" for i in range(1, 101)],
    "age": [20 + (i % 40) for i in range(100)],
    "ville": ["Paris", "Lyon", "Marseille", "Toulouse", "Bordeaux"] * 20,
    "salaire": [30000 + (i * 500) for i in range(100)]
})

# ============ ENDPOINT JSON ============

@app.get("/data")
def get_data(
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    ville: str = None,
    min_salaire: int = None
):
    """Retourne les données avec pagination et filtres."""
    
    result = df.copy()
    
    # Appliquer les filtres
    if ville:
        result = result[result["ville"] == ville]
    if min_salaire:
        result = result[result["salaire"] >= min_salaire]
    
    # Pagination
    total = len(result)
    start = (page - 1) * page_size
    end = start + page_size
    
    return {
        "total": total,
        "page": page,
        "page_size": page_size,
        "pages": (total + page_size - 1) // page_size,
        "data": result.iloc[start:end].to_dict(orient="records")
    }

# ============ ENDPOINT CSV ============

@app.get("/data/export/csv")
def export_csv():
    """Exporte les données en CSV."""
    
    stream = io.StringIO()
    df.to_csv(stream, index=False)
    
    response = StreamingResponse(
        iter([stream.getvalue()]),
        media_type="text/csv"
    )
    response.headers["Content-Disposition"] = "attachment; filename=data.csv"
    
    return response

# ============ ENDPOINT AGRÉGATIONS ============

@app.get("/data/stats")
def get_stats():
    """Retourne des statistiques agrégées."""
    
    stats = {
        "total_users": len(df),
        "age_moyen": round(df["age"].mean(), 2),
        "salaire_moyen": round(df["salaire"].mean(), 2),
        "salaire_median": round(df["salaire"].median(), 2),
        "par_ville": df.groupby("ville").agg({
            "id": "count",
            "salaire": "mean"
        }).round(2).to_dict(orient="index")
    }
    
    return stats

# Démonstration : Conversion DataFrame → JSON
import pandas as pd

df = pd.DataFrame({
    "id": [1, 2, 3],
    "nom": ["Alice", "Bob", "Charlie"],
    "salaire": [50000, 60000, 55000]
})

# Différentes conversions
print("📊 DataFrame original:")
print(df)

print("\n📄 to_dict('records') → pour API JSON:")
print(df.to_dict(orient="records"))

print("\n📄 to_dict('index') → pour stats par clé:")
print(df.set_index('nom')['salaire'].to_dict())

5️⃣ Connexion à une base de données

Avec SQLAlchemy (PostgreSQL, MySQL, SQLite)

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# URL de connexion
DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"
# Pour SQLite: "sqlite:///./data.db"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Dependency pour injecter la session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# models.py
from sqlalchemy import Column, Integer, String, Float, DateTime
from database import Base
from datetime import datetime

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    nom = Column(String(100), nullable=False)
    email = Column(String(100), unique=True, index=True)
    age = Column(Integer)
    created_at = Column(DateTime, default=datetime.now)

# main.py
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from database import get_db
import models

app = FastAPI()

@app.get("/users")
def get_users(db: Session = Depends(get_db), skip: int = 0, limit: int = 10):
    users = db.query(models.User).offset(skip).limit(limit).all()
    return users

Avec MongoDB (PyMongo)

# mongodb_api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from pymongo import MongoClient
from bson import ObjectId
from typing import Optional

app = FastAPI(title="MongoDB API")

# Connexion MongoDB
client = MongoClient("mongodb://localhost:27017")
db = client["data_engineering"]
collection = db["pipelines"]

# Helper pour ObjectId
class PyObjectId(ObjectId):
    @classmethod
    def __get_validators__(cls):
        yield cls.validate
    
    @classmethod
    def validate(cls, v, field):
        if not ObjectId.is_valid(v):
            raise ValueError("Invalid ObjectId")
        return ObjectId(v)

# Modèle
class Pipeline(BaseModel):
    id: Optional[str] = Field(default=None, alias="_id")
    nom: str
    description: str
    schedule: str
    
    class Config:
        populate_by_name = True

# CRUD
@app.get("/pipelines")
def list_pipelines():
    pipelines = []
    for doc in collection.find():
        doc["_id"] = str(doc["_id"])
        pipelines.append(doc)
    return pipelines

@app.post("/pipelines")
def create_pipeline(pipeline: Pipeline):
    result = collection.insert_one(pipeline.model_dump(exclude={"id"}))
    return {"id": str(result.inserted_id)}

@app.get("/pipelines/{pipeline_id}")
def get_pipeline(pipeline_id: str):
    doc = collection.find_one({"_id": ObjectId(pipeline_id)})
    if not doc:
        raise HTTPException(status_code=404, detail="Pipeline non trouvé")
    doc["_id"] = str(doc["_id"])
    return doc

@app.delete("/pipelines/{pipeline_id}")
def delete_pipeline(pipeline_id: str):
    result = collection.delete_one({"_id": ObjectId(pipeline_id)})
    if result.deleted_count == 0:
        raise HTTPException(status_code=404, detail="Pipeline non trouvé")
    return {"deleted": True}

6️⃣ Fonctionnalités avancées

Background Tasks (tâches en arrière-plan)

Utile pour les webhooks et le déclenchement de pipelines.

from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

def run_etl_pipeline(pipeline_id: int):
    """Simule l'exécution d'un pipeline ETL."""
    print(f" Démarrage du pipeline {pipeline_id}")
    time.sleep(10)  # Simule un traitement long
    print(f" Pipeline {pipeline_id} terminé")

@app.post("/pipelines/{pipeline_id}/run")
def trigger_pipeline(pipeline_id: int, background_tasks: BackgroundTasks):
    """Déclenche un pipeline en arrière-plan."""
    
    # Ajoute la tâche en arrière-plan
    background_tasks.add_task(run_etl_pipeline, pipeline_id)
    
    # Répond immédiatement
    return {
        "message": f"Pipeline {pipeline_id} déclenché",
        "status": "running"
    }

Webhook pour déclencher un pipeline

@app.post("/webhook/data-arrival")
def webhook_data_arrival(
    payload: dict,
    background_tasks: BackgroundTasks
):
    """Webhook appelé quand de nouvelles données arrivent."""
    
    source = payload.get("source")
    file_path = payload.get("file_path")
    
    # Déclencher le pipeline approprié
    background_tasks.add_task(
        process_new_data, 
        source=source, 
        file_path=file_path
    )
    
    return {"status": "accepted", "message": "Pipeline déclenché"}

Upload de fichiers

Permet d'ingérer des fichiers (CSV, JSON, Parquet) via API.

from fastapi import FastAPI, UploadFile, File
import pandas as pd
import io

app = FastAPI()

@app.post("/upload/csv")
async def upload_csv(file: UploadFile = File(...)):
    """Upload et analyse un fichier CSV."""
    
    # Vérifier l'extension
    if not file.filename.endswith('.csv'):
        raise HTTPException(status_code=400, detail="Fichier CSV requis")
    
    # Lire le contenu
    contents = await file.read()
    
    # Charger dans Pandas
    df = pd.read_csv(io.StringIO(contents.decode('utf-8')))
    
    return {
        "filename": file.filename,
        "rows": len(df),
        "columns": list(df.columns),
        "dtypes": df.dtypes.astype(str).to_dict(),
        "preview": df.head(5).to_dict(orient="records")
    }

@app.post("/upload/batch")
async def upload_multiple(files: list[UploadFile] = File(...)):
    """Upload plusieurs fichiers."""
    
    results = []
    for file in files:
        results.append({
            "filename": file.filename,
            "size": len(await file.read())
        })
    
    return {"uploaded": len(files), "files": results}

Documentation automatique

FastAPI génère automatiquement une documentation interactive !

URLs de documentation

Personnaliser la documentation

from fastapi import FastAPI

app = FastAPI(
    title="Data Pipeline API",
    description="""
    ## API pour gérer les pipelines de données
    
    Cette API permet de :
    - Gérer les pipelines ETL
    - Consulter les données
    - Déclencher des exécutions
    
    ### Authentification
    Utilise un token Bearer dans le header `Authorization`.
    """,
    version="1.0.0",
    contact={
        "name": "Data Team",
        "email": "data@company.com"
    },
    license_info={
        "name": "MIT"
    }
)

# Documenter un endpoint
@app.get(
    "/pipelines",
    summary="Liste les pipelines",
    description="Retourne tous les pipelines avec pagination",
    response_description="Liste des pipelines",
    tags=["Pipelines"]
)
def list_pipelines():
    pass

7️⃣ Exemple complet : API Data Pipeline

Voici un fichier complet à copier et tester :

# pipeline_api.py
"""
API FastAPI pour Data Engineering
Lancer avec : uvicorn pipeline_api:app --reload
"""

from fastapi import FastAPI, HTTPException, Query, BackgroundTasks, UploadFile, File
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
import pandas as pd
import io
import time

# ============ APPLICATION ============

app = FastAPI(
    title="Data Pipeline API",
    description="API pour gérer et monitorer les pipelines de données",
    version="1.0.0"
)

# ============ MODÈLES ============

class PipelineCreate(BaseModel):
    nom: str = Field(..., min_length=3, max_length=100)
    description: Optional[str] = None
    schedule: str = Field(..., description="Expression cron")
    source: str = Field(..., description="Source de données")
    destination: str = Field(..., description="Destination")

class Pipeline(PipelineCreate):
    id: int
    status: str = "idle"
    last_run: Optional[datetime] = None
    created_at: datetime

class RunResult(BaseModel):
    pipeline_id: int
    status: str
    rows_processed: int
    duration_seconds: float
    timestamp: datetime

# ============ STORAGE (simulé) ============

pipelines_db: dict[int, dict] = {}
runs_history: list[dict] = []
next_id = 1

# ============ HELPERS ============

def simulate_pipeline_run(pipeline_id: int):
    """Simule l'exécution d'un pipeline."""
    import random
    
    pipeline = pipelines_db.get(pipeline_id)
    if not pipeline:
        return
    
    pipeline["status"] = "running"
    start = time.time()
    
    # Simulation
    time.sleep(random.uniform(1, 3))
    rows = random.randint(1000, 100000)
    
    duration = time.time() - start
    pipeline["status"] = "idle"
    pipeline["last_run"] = datetime.now()
    
    # Enregistrer le run
    runs_history.append({
        "pipeline_id": pipeline_id,
        "status": "success",
        "rows_processed": rows,
        "duration_seconds": round(duration, 2),
        "timestamp": datetime.now()
    })

# ============ ENDPOINTS ============

@app.get("/", tags=["Health"])
def root():
    return {"status": "healthy", "service": "Data Pipeline API"}

# --- CRUD Pipelines ---

@app.post("/pipelines", response_model=Pipeline, status_code=201, tags=["Pipelines"])
def create_pipeline(pipeline: PipelineCreate):
    """Crée un nouveau pipeline."""
    global next_id
    
    new_pipeline = {
        "id": next_id,
        **pipeline.model_dump(),
        "status": "idle",
        "last_run": None,
        "created_at": datetime.now()
    }
    
    pipelines_db[next_id] = new_pipeline
    next_id += 1
    
    return new_pipeline

@app.get("/pipelines", response_model=list[Pipeline], tags=["Pipelines"])
def list_pipelines(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    status: Optional[str] = None
):
    """Liste tous les pipelines."""
    pipelines = list(pipelines_db.values())
    
    if status:
        pipelines = [p for p in pipelines if p["status"] == status]
    
    return pipelines[skip:skip + limit]

@app.get("/pipelines/{pipeline_id}", response_model=Pipeline, tags=["Pipelines"])
def get_pipeline(pipeline_id: int):
    """Récupère un pipeline par ID."""
    if pipeline_id not in pipelines_db:
        raise HTTPException(status_code=404, detail="Pipeline non trouvé")
    return pipelines_db[pipeline_id]

@app.delete("/pipelines/{pipeline_id}", status_code=204, tags=["Pipelines"])
def delete_pipeline(pipeline_id: int):
    """Supprime un pipeline."""
    if pipeline_id not in pipelines_db:
        raise HTTPException(status_code=404, detail="Pipeline non trouvé")
    del pipelines_db[pipeline_id]

# --- Exécution ---

@app.post("/pipelines/{pipeline_id}/run", tags=["Execution"])
def run_pipeline(pipeline_id: int, background_tasks: BackgroundTasks):
    """Déclenche l'exécution d'un pipeline."""
    if pipeline_id not in pipelines_db:
        raise HTTPException(status_code=404, detail="Pipeline non trouvé")
    
    pipeline = pipelines_db[pipeline_id]
    if pipeline["status"] == "running":
        raise HTTPException(status_code=409, detail="Pipeline déjà en cours")
    
    background_tasks.add_task(simulate_pipeline_run, pipeline_id)
    
    return {"message": f"Pipeline {pipeline_id} démarré", "status": "running"}

@app.get("/pipelines/{pipeline_id}/runs", response_model=list[RunResult], tags=["Execution"])
def get_pipeline_runs(pipeline_id: int, limit: int = 10):
    """Historique des exécutions d'un pipeline."""
    runs = [r for r in runs_history if r["pipeline_id"] == pipeline_id]
    return runs[-limit:]

# --- Data ---

@app.post("/upload", tags=["Data"])
async def upload_data(file: UploadFile = File(...)):
    """Upload un fichier de données."""
    contents = await file.read()
    
    if file.filename.endswith('.csv'):
        df = pd.read_csv(io.StringIO(contents.decode('utf-8')))
    elif file.filename.endswith('.json'):
        df = pd.read_json(io.StringIO(contents.decode('utf-8')))
    else:
        raise HTTPException(status_code=400, detail="Format non supporté")
    
    return {
        "filename": file.filename,
        "rows": len(df),
        "columns": list(df.columns),
        "preview": df.head().to_dict(orient="records")
    }

# ============ MAIN ============

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

🚀 Lancer et tester

# Terminal 1 : Lancer le serveur
uvicorn pipeline_api:app --reload

# Terminal 2 : Tester avec curl

# Créer un pipeline
curl -X POST "http://localhost:8000/pipelines" \
  -H "Content-Type: application/json" \
  -d '{"nom": "ETL Users", "schedule": "0 * * * *", "source": "postgres", "destination": "bigquery"}'

# Lister les pipelines
curl "http://localhost:8000/pipelines"

# Déclencher un pipeline
curl -X POST "http://localhost:8000/pipelines/1/run"

# Voir l'historique
curl "http://localhost:8000/pipelines/1/runs"

8️⃣ Bonnes pratiques

Structure de projet recommandée

my_api/
├── app/
│   ├── __init__.py
│   ├── main.py              # Point d'entrée FastAPI
│   ├── config.py            # Configuration (env vars)
│   ├── database.py          # Connexion DB
│   ├── models/              # Modèles SQLAlchemy/Pydantic
│   │   ├── __init__.py
│   │   └── pipeline.py
│   ├── schemas/             # Schémas Pydantic (request/response)
│   │   ├── __init__.py
│   │   └── pipeline.py
│   ├── routers/             # Endpoints par domaine
│   │   ├── __init__.py
│   │   ├── pipelines.py
│   │   └── data.py
│   └── services/            # Logique métier
│       ├── __init__.py
│       └── etl.py
├── tests/
│   ├── __init__.py
│   └── test_pipelines.py
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── README.md

Sécurité

# Ne jamais hardcoder les secrets !
import os
from dotenv import load_dotenv

load_dotenv()

DATABASE_URL = os.getenv("DATABASE_URL")
API_KEY = os.getenv("API_KEY")

Checklist production

Résumé

Ce que tu as appris

Cas d'usage Data Engineering

Commandes essentielles

# Installation
pip install fastapi uvicorn

# Lancer en dev
uvicorn main:app --reload

# Lancer en prod
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

📚 Ressources

Documentation officielle

• FastAPI Documentation — Excellente et complète
• Pydantic Documentation
• Uvicorn Documentation

Tutoriels

• FastAPI Tutorial — Tutoriel officiel
• Real Python - FastAPI
• TestDriven.io - FastAPI

Outils complémentaires

• SQLModel — ORM par le créateur de FastAPI
• Strawberry — GraphQL avec FastAPI
• Celery — Tâches asynchrones avancées

🎉 Félicitations ! Tu maîtrises maintenant FastAPI pour le Data Engineering.