Connexion base de données (SQLAlchemy)

Ton CRUD fonctionne, mais les données disparaissent à chaque redémarrage — un dictionnaire Python n’est pas une base de données. Ce chapitre te connecte à PostgreSQL avec SQLAlchemy, l’ORM le plus utilisé de l’écosystème Python. Tu vas structurer ton projet en couches propres et testables, injecter la session DB avec le système Depends de FastAPI, gérer les relations entre tables, et ajouter les middleware indispensables en production.

Architecture du projet

Avant de coder, la structure. Chaque fichier a un rôle précis — c’est cette séparation qui rend le projet maintenable à 6 mois :

project/
├── main.py          # Endpoints FastAPI
├── database.py      # Connexion et session DB
├── models.py        # Tables SQLAlchemy (ORM)
├── schemas.py       # Modèles Pydantic (validation)
└── crud.py          # Fonctions d'accès aux données

💡 Cette architecture n’est pas du perfectionnisme. C’est ce qui te permet de tester les fonctions CRUD sans démarrer l’API, de changer de base de données sans toucher aux endpoints, de faire des code reviews lisibles, et d’onboarder un nouveau développeur en 10 minutes au lieu d’une journée.

Connexion à PostgreSQL

Lance PostgreSQL en une commande Docker, puis configure la connexion :

# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql://fastapi:secret123@localhost:5432/app"

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

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

get_db() est un générateur — FastAPI crée une session au début de la requête et la ferme automatiquement à la fin, même en cas d’erreur. C’est le pattern d’injection de dépendances qui remplace les db.close() oubliés partout.

⚠️ Ne mets jamais l’URL de la DB en dur en prod. Utilise os.getenv("DATABASE_URL") ou un fichier .env avec python-dotenv. Le mot de passe n’a rien à faire dans Git.

Modèles SQLAlchemy vs Pydantic

C’est la confusion classique : les modèles SQLAlchemy représentent les tables (côté DB), les schémas Pydantic valident les données (côté API). Les deux coexistent :

# models.py — ce qui est en base
from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from database import Base

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False)
    email = Column(String(255), unique=True, nullable=False, index=True)
    hashed_password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    articles = relationship("Article", back_populates="author")

class Article(Base):
    __tablename__ = "articles"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    published = Column(Boolean, default=False)
    author_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    author = relationship("User", back_populates="articles")

Les relationship() câblent les relations — user.articles renvoie la liste des articles d’un utilisateur, article.author renvoie l’auteur. SQLAlchemy génère les JOINs automatiquement.

La couche CRUD

Les fonctions CRUD isolent l’accès aux données. Les endpoints appellent ces fonctions sans connaître les détails SQL :

# crud.py
from sqlalchemy.orm import Session
from models import User, Article

def get_user(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()

def create_user(db: Session, user):
    db_user = User(name=user.name, email=user.email,
                   hashed_password=user.password + "_hashed")
    db.add(db_user)
    db.commit()
    db.refresh(db_user)  # recharge depuis la DB (id, created_at)
    return db_user

def create_article(db: Session, article, author_id: int):
    db_article = Article(**article.model_dump(), author_id=author_id)
    db.add(db_article)
    db.commit()
    db.refresh(db_article)
    return db_article

🔥 db.refresh() est essentiel après un commit(). Sans lui, l’objet Python ne contient pas les champs auto-générés (id, created_at). Tu renverrais un objet incomplet au client.

Assembler les couches dans main.py

Les endpoints injectent la session DB via Depends(get_db) :

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

models.Base.metadata.create_all(bind=engine)
app = FastAPI(title="Blog API")

@app.post("/users", response_model=schemas.UserResponse, status_code=201)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    if crud.get_user_by_email(db, user.email):
        raise HTTPException(400, "Email déjà utilisé")
    return crud.create_user(db, user)

@app.post("/users/{user_id}/articles", response_model=schemas.ArticleResponse, status_code=201)
def create_article(user_id: int, article: schemas.ArticleCreate, db: Session = Depends(get_db)):
    if not crud.get_user(db, user_id):
        raise HTTPException(404, "Utilisateur non trouvé")
    return crud.create_article(db, article, author_id=user_id)

Depends gère le cycle de vie de la session : création → injection → fermeture. Pas de risque de fuite de connexions.

Middleware et CORS

Les middleware s’exécutent avant et après chaque requête — parfaits pour le monitoring :

import time, logging
logger = logging.getLogger("api")

@app.middleware("http")
async def log_requests(request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    logger.info(f"{request.method} {request.url.path} → {response.status_code} ({duration:.3f}s)")
    return response

Si ton frontend est sur un domaine différent de l’API, le navigateur bloque les requêtes. CORS autorise les origines :

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://app.devopslab.ch"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

🎯 Jamais allow_origins=["*"] en production. Ça autorise n’importe quel site à appeler ton API avec les cookies de l’utilisateur. Liste explicitement les domaines autorisés.

Les pièges classiques

Sessions qui fuient. Si tu crées une SessionLocal() manuellement sans finally: db.close(), chaque requête consomme une connexion du pool sans la libérer. Après 100 requêtes, PostgreSQL refuse les nouvelles connexions. Utilise toujours Depends(get_db).

create_all() en production. Base.metadata.create_all() crée les tables manquantes mais ne modifie jamais les tables existantes. Si tu ajoutes une colonne, elle n’apparaîtra pas. En production, utilise Alembic pour les migrations de schéma — c’est le Git de ta base de données.

Le N+1 query. Tu charges 100 articles, puis pour chaque article tu fais une requête pour charger l’auteur → 101 requêtes au lieu d’une. Solution : joinedload de SQLAlchemy pour charger les relations en une seule requête.

ORM vs SQL brut. L’ORM couvre 90% des cas. Pour les requêtes d’agrégation complexes, les window functions ou les opérations JSONB, passe au SQL brut avec db.execute(text("SELECT ...")). Ne force pas l’ORM quand le SQL est plus lisible.

autocommit=False et les rollbacks. Avec autocommit=False (recommandé), chaque db.commit() est explicite. Si une erreur survient, appelle db.rollback() avant de réessayer ou de renvoyer une erreur — sinon la session reste dans un état corrompu et les requêtes suivantes échouent.

⚠️ Gère les IntegrityError. Si deux requêtes créent un utilisateur avec le même email simultanément, SQLAlchemy lève une IntegrityError. Attrape-la, fais un db.rollback(), et renvoie un 409 Conflict propre. Ce pattern est indispensable en production :

from sqlalchemy.exc import IntegrityError

@app.post("/users", status_code=201)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    try:
        return crud.create_user(db, user)
    except IntegrityError:
        db.rollback()
        raise HTTPException(409, "Email déjà utilisé")

Ce qu’on retient

🎯 SQLAlchemy + Depends = une couche de données propre, testable et sans fuites.

Les essentiels :

• 5 fichiers — database.py, models.py, schemas.py, crud.py, main.py
• Depends(get_db) — injection de session avec fermeture automatique
• Modèles SQLAlchemy ≠ schémas Pydantic — DB vs validation, deux rôles distincts
• db.refresh() — recharge les champs auto-générés après commit
• CORS — liste explicite des origines en prod, jamais de wildcard
• Alembic — pour les migrations de schéma, create_all() ne suffit pas

Prochain chapitre : tests et documentation — pytest, TestClient, et les bonnes pratiques pour une API documentée et fiable. 🚀