- CRUD FastAPI Python & Angular : REST API complète

1. Architecture et prérequis

Une application full-stack moderne nécessite une séparation claire des responsabilités. FastAPI (Python) gère la logique métier et l'accès aux données côté serveur, tandis qu'Angular orchestre l'interface utilisateur côté client. Les deux communiquent via une REST API JSON standardisée.

Architecture globale du projet

Voici la structure que nous allons construire, typique d'une application CRUD de production :

# Structure du projet
my-fullstack-app/
├── backend/                    # API FastAPI Python
│   ├── main.py                 # Point d'entrée de l'application
│   ├── database.py             # Configuration SQLAlchemy
│   ├── models.py               # Modèles ORM (tables BDD)
│   ├── schemas.py              # Schémas Pydantic (validation)
│   ├── crud.py                 # Fonctions CRUD (logique BDD)
│   ├── routers/
│   │   └── items.py            # Routes API pour la ressource Item
│   ├── requirements.txt        # Dépendances Python
│   └── .env                    # Variables d'environnement (non commité)
│
└── frontend/                   # Application Angular
    ├── src/
    │   ├── app/
    │   │   ├── models/
    │   │   │   └── item.model.ts       # Interface TypeScript
    │   │   ├── services/
    │   │   │   └── item.service.ts     # Service HttpClient
    │   │   ├── components/
    │   │   │   ├── item-list/          # Composant liste
    │   │   │   └── item-form/          # Composant formulaire
    │   │   └── app.module.ts
    │   └── environments/
    │       └── environment.ts          # URL de l'API
    └── package.json

Prérequis techniques

2. Installation et configuration FastAPI

Commençons par mettre en place l'environnement backend Python avec toutes les dépendances nécessaires à une API de production.

Création de l'environnement virtuel Python

# Créer le dossier backend et l'environnement virtuel
mkdir backend && cd backend
python -m venv venv

# Activer l'environnement virtuel
source venv/bin/activate        # Linux/macOS
venv\Scripts\activate           # Windows

# Installer les dépendances
pip install fastapi uvicorn[standard] sqlalchemy psycopg2-binary pydantic-settings python-dotenv alembic

Fichier requirements.txt

# requirements.txt — Dépendances du projet FastAPI
fastapi==0.110.0            # Framework API principal
uvicorn[standard]==0.29.0   # Serveur ASGI performant
sqlalchemy==2.0.29          # ORM Python moderne
psycopg2-binary==2.9.9      # Driver PostgreSQL
pydantic-settings==2.2.1    # Gestion config avec variables d'env
python-dotenv==1.0.1        # Chargement fichier .env
alembic==1.13.1             # Migrations de base de données
python-jose[cryptography]==3.3.0  # JWT (pour sécurité future)

Variables d'environnement (.env)

# .env — NE PAS commiter ce fichier !
# Copier depuis .env.example et remplir les valeurs

DATABASE_URL=postgresql://myuser:mypassword@localhost:5432/myapp_db
SECRET_KEY=une-clé-secrète-très-longue-et-aléatoire
ENVIRONMENT=development
ALLOWED_ORIGINS=http://localhost:4200

Configuration de l'application (main.py)

# main.py — Point d'entrée de l'application FastAPI
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
import os

from database import engine, Base
from routers import items

# Gestionnaire de cycle de vie de l'application
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Démarrage : créer les tables si elles n'existent pas
    Base.metadata.create_all(bind=engine)
    yield
    # Arrêt : nettoyage éventuel des ressources

# Création de l'instance FastAPI avec métadonnées
app = FastAPI(
    title="Mon API CRUD",
    description="API REST complète avec FastAPI, SQLAlchemy et PostgreSQL",
    version="1.0.0",
    lifespan=lifespan
)

# Origines autorisées pour Angular (lues depuis l'env)
origins = os.getenv("ALLOWED_ORIGINS", "http://localhost:4200").split(",")

# Configuration du middleware CORS — obligatoire pour Angular
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,       # Domaines autorisés
    allow_credentials=True,      # Cookies cross-origin autorisés
    allow_methods=["*"],         # GET, POST, PUT, DELETE, OPTIONS
    allow_headers=["*"],         # Tous les headers HTTP
)

# Inclusion des routeurs avec préfixe et tags pour la doc
app.include_router(items.router, prefix="/api/v1/items", tags=["Items"])

# Route de santé pour le monitoring
@app.get("/health", tags=["Health"])
async def health_check():
    return {"status": "ok", "version": "1.0.0"}

Lancer le serveur de développement

# Démarrer FastAPI avec rechargement automatique
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# Accéder à la documentation Swagger automatique
# http://localhost:8000/docs

# Documentation alternative ReDoc
# http://localhost:8000/redoc

3. Modèles SQLAlchemy et schémas Pydantic

FastAPI s'appuie sur deux couches de définition des données distinctes : SQLAlchemy pour la persistance en base de données et Pydantic pour la validation et la sérialisation des données JSON. Cette séparation permet une architecture propre et maintenable.

Configuration de la base de données (database.py)

# database.py — Configuration SQLAlchemy et session
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os
from dotenv import load_dotenv

load_dotenv()  # Charger les variables d'environnement depuis .env

# URL de connexion PostgreSQL depuis l'environnement
DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql://postgres:postgres@localhost:5432/myapp_db"
)

# Créer le moteur SQLAlchemy
# pool_pre_ping=True : vérifie les connexions avant utilisation
engine = create_engine(DATABASE_URL, pool_pre_ping=True)

# Factory de sessions — chaque requête HTTP obtient sa propre session
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Classe de base pour tous les modèles SQLAlchemy
Base = declarative_base()

# Dépendance FastAPI pour injecter la session dans les routes
def get_db():
    db = SessionLocal()
    try:
        yield db       # Fournit la session à la route
    finally:
        db.close()    # Ferme toujours la session après usage

Modèle SQLAlchemy (models.py)

# models.py — Modèles ORM SQLAlchemy (structure des tables)
from sqlalchemy import Column, Integer, String, Boolean, DateTime, Text
from sqlalchemy.sql import func
from database import Base

class Item(Base):
    """Modèle SQLAlchemy représentant la table 'items' en base."""
    __tablename__ = "items"

    id          = Column(Integer, primary_key=True, index=True)
    title       = Column(String(200), nullable=False, index=True)
    description = Column(Text, nullable=True)
    price       = Column(Integer, nullable=False, default=0)
    is_active   = Column(Boolean, default=True, nullable=False)

    # Timestamps automatiques gérés par la base de données
    created_at  = Column(DateTime(timezone=True), server_default=func.now())
    updated_at  = Column(DateTime(timezone=True), onupdate=func.now())

Schémas Pydantic (schemas.py)

# schemas.py — Schémas Pydantic pour validation et sérialisation
from pydantic import BaseModel, Field, ConfigDict
from datetime import datetime
from typing import Optional

# Schéma de base avec les champs communs (sans id ni timestamps)
class ItemBase(BaseModel):
    title: str = Field(
        ...,
        min_length=1,
        max_length=200,
        description="Titre de l'article (obligatoire)"
    )
    description: Optional[str] = Field(
        default=None,
        description="Description optionnelle"
    )
    price: int = Field(
        default=0,
        ge=0,               # Valeur >= 0 (validation automatique)
        description="Prix en centimes"
    )
    is_active: bool = Field(default=True)

# Schéma pour la création (POST) — hérite de ItemBase
class ItemCreate(ItemBase):
    pass  # Tous les champs de ItemBase, sans modification

# Schéma pour la mise à jour (PUT) — tous les champs optionnels
class ItemUpdate(BaseModel):
    title:       Optional[str] = Field(default=None, min_length=1, max_length=200)
    description: Optional[str] = None
    price:       Optional[int] = Field(default=None, ge=0)
    is_active:   Optional[bool] = None

# Schéma de réponse complet (GET) — inclut id et timestamps
class ItemResponse(ItemBase):
    id:         int
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None

    # Configuration pour lire depuis les attributs SQLAlchemy (ORM)
    model_config = ConfigDict(from_attributes=True)

# Schéma pour la liste paginée
class ItemListResponse(BaseModel):
    items: list[ItemResponse]
    total: int
    page:  int
    size:  int

4. Endpoints CRUD complets

Maintenant que la structure de données est définie, nous allons implémenter les opérations CRUD en deux fichiers : crud.py pour la logique base de données et routers/items.py pour les routes HTTP.

Couche d'accès aux données (crud.py)

# crud.py — Fonctions CRUD SQLAlchemy (séparées des routes HTTP)
from sqlalchemy.orm import Session
from sqlalchemy import func
from typing import Optional
import models
import schemas

def get_item(db: Session, item_id: int) -> Optional[models.Item]:
    """Récupère un item par son ID, retourne None si introuvable."""
    return db.query(models.Item).filter(models.Item.id == item_id).first()

def get_items(
    db: Session,
    skip: int = 0,
    limit: int = 10,
    active_only: bool = False
) -> tuple[list[models.Item], int]:
    """
    Récupère une liste paginée d'items.
    Retourne (items, total) pour la pagination côté Angular.
    """
    query = db.query(models.Item)

    # Filtre optionnel sur les items actifs
    if active_only:
        query = query.filter(models.Item.is_active == True)

    total = query.count()                        # Compte total pour pagination
    items = query.offset(skip).limit(limit).all() # Page demandée
    return items, total

def create_item(db: Session, item: schemas.ItemCreate) -> models.Item:
    """Crée un nouvel item en base de données."""
    db_item = models.Item(**item.model_dump())   # Convertit le schéma en modèle
    db.add(db_item)
    db.commit()
    db.refresh(db_item)                          # Recharge depuis la BDD (récupère id, timestamps)
    return db_item

def update_item(
    db: Session,
    item_id: int,
    item_update: schemas.ItemUpdate
) -> Optional[models.Item]:
    """
    Met à jour partiellement un item (PATCH sémantique).
    Seuls les champs fournis sont modifiés.
    """
    db_item = get_item(db, item_id)
    if db_item is None:
        return None

    # Exclure les valeurs None (champs non fournis dans la requête)
    update_data = item_update.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(db_item, field, value)

    db.commit()
    db.refresh(db_item)
    return db_item

def delete_item(db: Session, item_id: int) -> bool:
    """Supprime un item. Retourne True si supprimé, False si introuvable."""
    db_item = get_item(db, item_id)
    if db_item is None:
        return False
    db.delete(db_item)
    db.commit()
    return True

def search_items(db: Session, query: str, limit: int = 10) -> list[models.Item]:
    """Recherche des items par titre (recherche insensible à la casse)."""
    return (
        db.query(models.Item)
        .filter(models.Item.title.ilike(f"%{query}%"))
        .limit(limit)
        .all()
    )

Routeur FastAPI (routers/items.py)

# routers/items.py — Routes HTTP pour la ressource Item
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from database import get_db
import schemas
import crud

# APIRouter : groupe de routes avec préfixe défini dans main.py
router = APIRouter()

@router.get("/", response_model=schemas.ItemListResponse)
def read_items(
    page: int = Query(default=1, ge=1, description="Numéro de page"),
    size: int = Query(default=10, ge=1, le=100, description="Items par page"),
    active_only: bool = Query(default=False),
    db: Session = Depends(get_db)   # Injection de la session BDD
):
    """Récupère la liste paginée des items."""
    skip = (page - 1) * size   # Calcul de l'offset depuis la page
    items, total = crud.get_items(db, skip=skip, limit=size, active_only=active_only)
    return schemas.ItemListResponse(
        items=items,
        total=total,
        page=page,
        size=size
    )

@router.get("/{item_id}", response_model=schemas.ItemResponse)
def read_item(item_id: int, db: Session = Depends(get_db)):
    """Récupère un item par son ID."""
    db_item = crud.get_item(db, item_id)
    if db_item is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item avec l'ID {item_id} introuvable"
        )
    return db_item

@router.post("/", response_model=schemas.ItemResponse, status_code=status.HTTP_201_CREATED)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    """Crée un nouvel item. Retourne 201 Created avec l'item créé."""
    return crud.create_item(db, item)

@router.put("/{item_id}", response_model=schemas.ItemResponse)
def update_item(
    item_id: int,
    item_update: schemas.ItemUpdate,
    db: Session = Depends(get_db)
):
    """Met à jour partiellement un item existant."""
    db_item = crud.update_item(db, item_id, item_update)
    if db_item is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item avec l'ID {item_id} introuvable"
        )
    return db_item

@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int, db: Session = Depends(get_db)):
    """Supprime un item. Retourne 204 No Content si succès."""
    deleted = crud.delete_item(db, item_id)
    if not deleted:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item avec l'ID {item_id} introuvable"
        )

@router.get("/search/", response_model=list[schemas.ItemResponse])
def search_items(
    q: str = Query(..., min_length=2, description="Terme de recherche"),
    db: Session = Depends(get_db)
):
    """Recherche des items par titre (insensible à la casse)."""
    return crud.search_items(db, q)

5. CORS et sécurité API

La sécurité d'une API publique commence par une configuration CORS rigoureuse et une validation stricte des entrées. FastAPI facilite les deux grâce à son middleware intégré et à Pydantic.

Configuration CORS avancée

# Configuration CORS différenciée par environnement
import os
from fastapi.middleware.cors import CORSMiddleware

ENVIRONMENT = os.getenv("ENVIRONMENT", "development")

if ENVIRONMENT == "production":
    # En production : domaines stricts uniquement
    allowed_origins = [
        "https://myapp.com",
        "https://www.myapp.com",
    ]
else:
    # En développement : Angular local autorisé
    allowed_origins = [
        "http://localhost:4200",   # Angular CLI
        "http://127.0.0.1:4200",
    ]

app.add_middleware(
    CORSMiddleware,
    allow_origins=allowed_origins,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["Content-Type", "Authorization", "X-Requested-With"],
    max_age=3600,              # Cache preflight 1 heure
)

Middleware de logging et rate limiting

# Middleware personnalisé pour logger les requêtes
from fastapi import Request
import time
import logging

logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    """Log chaque requête HTTP avec timing pour monitoring."""
    start_time = time.time()
    response = await call_next(request)
    process_time = (time.time() - start_time) * 1000

    logger.info(
        f"{request.method} {request.url.path} "
        f"→ {response.status_code} ({process_time:.1f}ms)"
    )
    # Header de timing pour le débogage front-end
    response.headers["X-Process-Time"] = f"{process_time:.1f}ms"
    return response

6. Service Angular avec HttpClient

Côté Angular, toute communication avec l'API FastAPI passe par un service dédié. Ce pattern centralise la logique HTTP, facilite les tests et évite la duplication de code dans les composants.

Interface TypeScript du modèle (item.model.ts)

// src/app/models/item.model.ts
// Miroir TypeScript des schémas Pydantic FastAPI

export interface Item {
  id: number;
  title: string;
  description: string | null;
  price: number;
  is_active: boolean;
  created_at: string | null;
  updated_at: string | null;
}

// Payload pour la création — sans id ni timestamps
export interface ItemCreate {
  title: string;
  description?: string | null;
  price: number;
  is_active?: boolean;
}

// Payload pour la mise à jour — tous les champs optionnels
export interface ItemUpdate {
  title?: string;
  description?: string | null;
  price?: number;
  is_active?: boolean;
}

// Réponse de liste paginée
export interface ItemListResponse {
  items: Item[];
  total: number;
  page: number;
  size: number;
}

Configuration de l'URL API (environment.ts)

// src/environments/environment.ts
export const environment = {
  production: false,
  apiUrl: 'http://localhost:8000/api/v1'  // URL FastAPI locale
};

// src/environments/environment.prod.ts
export const environment = {
  production: true,
  apiUrl: 'https://api.myapp.com/api/v1'  // URL FastAPI production
};

Service CRUD Angular (item.service.ts)

// src/app/services/item.service.ts
import { Injectable } from '@angular/core';
import { HttpClient, HttpParams } from '@angular/common/http';
import { Observable } from 'rxjs';
import { environment } from '../../environments/environment';
import { Item, ItemCreate, ItemUpdate, ItemListResponse } from '../models/item.model';

@Injectable({
  providedIn: 'root'  // Service singleton disponible dans toute l'app
})
export class ItemService {
  private readonly apiUrl = `${environment.apiUrl}/items`;

  constructor(private http: HttpClient) {}

  /**
   * Récupère la liste paginée des items.
   * @param page Numéro de page (commence à 1)
   * @param size Nombre d'items par page
   */
  getItems(page = 1, size = 10, activeOnly = false): Observable {
    const params = new HttpParams()
      .set('page', page.toString())
      .set('size', size.toString())
      .set('active_only', activeOnly.toString());

    return this.http.get(this.apiUrl, { params });
  }

  /** Récupère un item par son ID. */
  getItem(id: number): Observable {
    return this.http.get(`${this.apiUrl}/${id}`);
  }

  /** Crée un nouvel item. FastAPI retourne 201 avec l'item créé. */
  createItem(item: ItemCreate): Observable {
    return this.http.post(this.apiUrl + '/', item);
  }

  /** Met à jour partiellement un item existant. */
  updateItem(id: number, item: ItemUpdate): Observable {
    return this.http.put(`${this.apiUrl}/${id}`, item);
  }

  /** Supprime un item. FastAPI retourne 204 No Content. */
  deleteItem(id: number): Observable {
    return this.http.delete(`${this.apiUrl}/${id}`);
  }

  /** Recherche des items par titre. */
  searchItems(query: string): Observable {
    const params = new HttpParams().set('q', query);
    return this.http.get(`${this.apiUrl}/search/`, { params });
  }
}

Configuration HttpClient dans app.module.ts

// src/app/app.module.ts
import { NgModule } from '@angular/core';
import { BrowserModule } from '@angular/platform-browser';
import { HttpClientModule } from '@angular/common/http';
import { ReactiveFormsModule } from '@angular/forms';  // Pour les formulaires
import { AppRoutingModule } from './app-routing.module';
import { AppComponent } from './app.component';
import { ItemListComponent } from './components/item-list/item-list.component';
import { ItemFormComponent } from './components/item-form/item-form.component';

@NgModule({
  declarations: [AppComponent, ItemListComponent, ItemFormComponent],
  imports: [
    BrowserModule,
    AppRoutingModule,
    HttpClientModule,       // Nécessaire pour HttpClient
    ReactiveFormsModule     // Nécessaire pour les formulaires réactifs
  ],
  bootstrap: [AppComponent]
})
export class AppModule {}

7. Composants Angular : liste et formulaires

Les composants Angular orchestrent l'affichage et les interactions utilisateur. Nous allons créer deux composants : ItemListComponent pour afficher et gérer la liste, et ItemFormComponent pour la création et l'édition.

Composant liste avec pagination (item-list.component.ts)

// src/app/components/item-list/item-list.component.ts
import { Component, OnInit, OnDestroy } from '@angular/core';
import { Subject } from 'rxjs';
import { takeUntil, debounceTime, distinctUntilChanged } from 'rxjs/operators';
import { FormControl } from '@angular/forms';
import { ItemService } from '../../services/item.service';
import { Item, ItemListResponse } from '../../models/item.model';

@Component({
  selector: 'app-item-list',
  templateUrl: './item-list.component.html',
})
export class ItemListComponent implements OnInit, OnDestroy {
  items: Item[] = [];
  total = 0;
  currentPage = 1;
  pageSize = 10;
  loading = false;
  errorMessage = '';
  selectedItem: Item | null = null;

  // FormControl pour la barre de recherche avec debounce
  searchControl = new FormControl('');
  private destroy$ = new Subject();

  constructor(private itemService: ItemService) {}

  ngOnInit(): void {
    this.loadItems();

    // Recherche avec debounce 400ms — évite trop de requêtes API
    this.searchControl.valueChanges.pipe(
      debounceTime(400),
      distinctUntilChanged(),     // Ignore si valeur identique
      takeUntil(this.destroy$)
    ).subscribe(query => {
      if (query && query.length >= 2) {
        this.searchItems(query);
      } else if (!query) {
        this.loadItems();          // Retour à la liste complète si vide
      }
    });
  }

  loadItems(): void {
    this.loading = true;
    this.errorMessage = '';

    this.itemService
      .getItems(this.currentPage, this.pageSize)
      .pipe(takeUntil(this.destroy$))
      .subscribe({
        next: (response: ItemListResponse) => {
          this.items = response.items;
          this.total = response.total;
          this.loading = false;
        },
        error: (err) => {
          this.errorMessage = 'Impossible de charger les items.';
          this.loading = false;
          console.error('Erreur chargement items:', err);
        }
      });
  }

  searchItems(query: string): void {
    this.loading = true;
    this.itemService
      .searchItems(query)
      .pipe(takeUntil(this.destroy$))
      .subscribe({
        next: (items) => { this.items = items; this.loading = false; },
        error: () => { this.errorMessage = 'Erreur lors de la recherche.'; this.loading = false; }
      });
  }

  onDelete(id: number): void {
    if (!confirm('Supprimer cet item ?')) return;

    this.itemService.deleteItem(id)
      .pipe(takeUntil(this.destroy$))
      .subscribe({
        next: () => this.loadItems(),   // Recharge la liste après suppression
        error: () => { this.errorMessage = 'Impossible de supprimer l\'item.'; }
      });
  }

  onPageChange(page: number): void {
    this.currentPage = page;
    this.loadItems();
  }

  onEditItem(item: Item): void {
    this.selectedItem = { ...item };  // Clone pour éviter la mutation directe
  }

  onFormSaved(): void {
    this.selectedItem = null;
    this.loadItems();                 // Recharge après création/modification
  }

  // Nettoyage des subscriptions à la destruction du composant
  ngOnDestroy(): void {
    this.destroy$.next();
    this.destroy$.complete();
  }
}

Template liste HTML (item-list.component.html)

<!-- src/app/components/item-list/item-list.component.html -->
<div class="container mt-4">
  <div class="d-flex justify-content-between align-items-center mb-3">
    <h2>Gestion des items</h2>
    <button class="btn btn-primary" (click)="onEditItem(null)">
      + Nouvel item
    </button>
  </div>

  <!-- Barre de recherche -->
  <input [formControl]="searchControl" class="form-control mb-3"
         placeholder="Rechercher un item..." type="search">

  <!-- Message d'erreur -->
  <div *ngIf="errorMessage" class="alert alert-danger">
    {{ errorMessage }}
  </div>

  <!-- Indicateur de chargement -->
  <div *ngIf="loading" class="text-center py-4">
    <div class="spinner-border text-primary"></div>
  </div>

  <!-- Table des items -->
  <div class="table-responsive" *ngIf="!loading">
    <table class="table table-hover">
      <thead class="table-dark">
        <tr>
          <th>#</th>
          <th>Titre</th>
          <th>Prix</th>
          <th>Statut</th>
          <th>Actions</th>
        </tr>
      </thead>
      <tbody>
        <tr *ngFor="let item of items">
          <td>{{ item.id }}</td>
          <td>{{ item.title }}</td>
          <td>{{ item.price / 100 | currency:'EUR' }}</td>
          <td>
            <span [class]="item.is_active ? 'badge bg-success' : 'badge bg-secondary'">
              {{ item.is_active ? 'Actif' : 'Inactif' }}
            </span>
          </td>
          <td>
            <button class="btn btn-sm btn-outline-primary me-1" (click)="onEditItem(item)">
              Modifier
            </button>
            <button class="btn btn-sm btn-outline-danger" (click)="onDelete(item.id)">
              Supprimer
            </button>
          </td>
        </tr>
        <tr *ngIf="items.length === 0">
          <td colspan="5" class="text-center text-muted">Aucun item trouvé.</td>
        </tr>
      </tbody>
    </table>
  </div>

  <!-- Pagination simple -->
  <div class="d-flex justify-content-between align-items-center mt-2">
    <span class="text-muted">{{ total }} items au total</span>
    <div class="btn-group">
      <button class="btn btn-outline-secondary btn-sm"
              [disabled]="currentPage === 1"
              (click)="onPageChange(currentPage - 1)">Précédent</button>
      <button class="btn btn-outline-secondary btn-sm"
              [disabled]="currentPage * pageSize >= total"
              (click)="onPageChange(currentPage + 1)">Suivant</button>
    </div>
  </div>

  <!-- Formulaire en-dessous si item sélectionné -->
  <app-item-form *ngIf="selectedItem !== undefined"
                 [item]="selectedItem"
                 (saved)="onFormSaved()"
                 (cancelled)="selectedItem = undefined">
  </app-item-form>
</div>

Composant formulaire réactif (item-form.component.ts)

// src/app/components/item-form/item-form.component.ts
import { Component, Input, Output, EventEmitter, OnChanges } from '@angular/core';
import { FormBuilder, FormGroup, Validators } from '@angular/forms';
import { ItemService } from '../../services/item.service';
import { Item } from '../../models/item.model';

@Component({
  selector: 'app-item-form',
  templateUrl: './item-form.component.html',
})
export class ItemFormComponent implements OnChanges {
  @Input() item: Item | null = null;   // null = création, Item = modification
  @Output() saved = new EventEmitter();
  @Output() cancelled = new EventEmitter();

  form: FormGroup;
  submitting = false;
  errorMessage = '';

  constructor(private fb: FormBuilder, private itemService: ItemService) {
    this.form = this.fb.group({
      title: ['', [Validators.required, Validators.maxLength(200)]],
      description: [''],
      price: [0, [Validators.required, Validators.min(0)]],
      is_active: [true]
    });
  }

  // Réagit quand l'input @Input() item change
  ngOnChanges(): void {
    if (this.item) {
      this.form.patchValue({        // Pré-rempli le formulaire en mode édition
        title: this.item.title,
        description: this.item.description || '',
        price: this.item.price,
        is_active: this.item.is_active
      });
    } else {
      this.form.reset({ price: 0, is_active: true });  // Réinitialise en mode création
    }
  }

  onSubmit(): void {
    if (this.form.invalid) return;   // Bloque si validation échoue

    this.submitting = true;
    this.errorMessage = '';
    const formValue = this.form.value;

    const request$ = this.item
      ? this.itemService.updateItem(this.item.id, formValue)  // PUT si modification
      : this.itemService.createItem(formValue);               // POST si création

    request$.subscribe({
      next: () => {
        this.submitting = false;
        this.saved.emit();          // Notifie le parent que la sauvegarde est faite
      },
      error: (err) => {
        this.submitting = false;
        // FastAPI retourne des erreurs de validation détaillées en 422
        if (err.status === 422) {
          this.errorMessage = 'Données invalides. Vérifiez les champs.';
        } else {
          this.errorMessage = 'Une erreur est survenue. Réessayez.';
        }
      }
    });
  }

  get isEditing(): boolean {
    return this.item !== null;
  }
}

8. Gestion des erreurs et intercepteurs

Une application robuste gère les erreurs de façon cohérente. Les intercepteurs HTTP Angular permettent de centraliser la gestion des erreurs, des tokens d'authentification et du logging sans dupliquer le code dans chaque service.

Intercepteur d'erreurs globales (error.interceptor.ts)

// src/app/interceptors/error.interceptor.ts
import { Injectable } from '@angular/core';
import {
  HttpRequest, HttpHandler, HttpEvent,
  HttpInterceptor, HttpErrorResponse
} from '@angular/common/http';
import { Observable, throwError } from 'rxjs';
import { catchError, retry } from 'rxjs/operators';

@Injectable()
export class ErrorInterceptor implements HttpInterceptor {
  intercept(request: HttpRequest, next: HttpHandler): Observable> {
    return next.handle(request).pipe(
      // Réessayer automatiquement une fois pour les erreurs réseau
      retry({ count: 1, delay: 1000 }),
      catchError((error: HttpErrorResponse) => {
        let userMessage = 'Une erreur inattendue s\'est produite.';

        if (error.status === 0) {
          // Erreur réseau ou CORS — serveur injoignable
          userMessage = 'Impossible de joindre le serveur. Vérifiez votre connexion.';
        } else if (error.status === 404) {
          userMessage = 'La ressource demandée est introuvable.';
        } else if (error.status === 422) {
          // FastAPI retourne le détail des erreurs Pydantic dans error.error.detail
          const details = error.error?.detail;
          if (Array.isArray(details)) {
            userMessage = details.map((d: any) =>
              `${d.loc?.join('.')} : ${d.msg}`
            ).join(', ');
          }
        } else if (error.status >= 500) {
          userMessage = 'Erreur serveur. L\'équipe technique a été notifiée.';
        }

        // Enrichit l'erreur avec un message lisible par l'utilisateur
        return throwError(() => ({ ...error, userMessage }));
      })
    );
  }
}

// Enregistrer dans app.module.ts :
// { provide: HTTP_INTERCEPTORS, useClass: ErrorInterceptor, multi: true }

Gestion des erreurs côté FastAPI

# Gestionnaire d'exceptions global pour des erreurs JSON cohérentes
from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from sqlalchemy.exc import SQLAlchemyError

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """Retourne une réponse 422 structurée pour les erreurs Pydantic."""
    return JSONResponse(
        status_code=422,
        content={
            "message": "Erreur de validation",
            "detail": exc.errors()    # Détail de chaque champ invalide
        }
    )

@app.exception_handler(SQLAlchemyError)
async def database_exception_handler(request: Request, exc: SQLAlchemyError):
    """Capture les erreurs base de données et retourne un 500 propre."""
    logger.error(f"Erreur BDD: {exc}")
    return JSONResponse(
        status_code=500,
        content={"message": "Erreur interne du serveur."}
    )

9. Tests et déploiement production

Avant de déployer en production, des tests automatisés garantissent la fiabilité de l'API. FastAPI s'intègre parfaitement avec pytest et son client de test intégré.

Tests d'intégration FastAPI (test_items.py)

# tests/test_items.py — Tests d'intégration avec pytest
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from main import app
from database import Base, get_db

# Base SQLite en mémoire pour les tests (rapide, sans état persistant)
SQLALCHEMY_TEST_URL = "sqlite:///./test.db"
engine_test = create_engine(SQLALCHEMY_TEST_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(bind=engine_test)

def override_get_db():
    """Remplace la session BDD de prod par celle de test."""
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

# Surcharge la dépendance BDD pour les tests
app.dependency_overrides[get_db] = override_get_db

@pytest.fixture(autouse=True)
def setup_database():
    """Recrée les tables avant chaque test et les supprime après."""
    Base.metadata.create_all(bind=engine_test)
    yield
    Base.metadata.drop_all(bind=engine_test)

client = TestClient(app)

def test_create_item():
    """Test création d'un item valide — doit retourner 201."""
    response = client.post("/api/v1/items/", json={
        "title": "Item de test",
        "price": 1500,
        "is_active": True
    })
    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Item de test"
    assert data["id"] is not None

def test_get_items_empty():
    """Test liste vide — doit retourner total=0."""
    response = client.get("/api/v1/items/")
    assert response.status_code == 200
    assert response.json()["total"] == 0

def test_create_and_delete_item():
    """Test création puis suppression."""
    # Créer
    create_res = client.post("/api/v1/items/", json={"title": "À supprimer", "price": 0})
    item_id = create_res.json()["id"]

    # Supprimer
    delete_res = client.delete(f"/api/v1/items/{item_id}")
    assert delete_res.status_code == 204

    # Vérifier que l'item n'existe plus
    get_res = client.get(f"/api/v1/items/{item_id}")
    assert get_res.status_code == 404

def test_create_item_invalid_title():
    """Test création avec titre vide — Pydantic doit rejeter (422)."""
    response = client.post("/api/v1/items/", json={"title": "", "price": 0})
    assert response.status_code == 422

Dockerfile multi-stage pour la production

# Dockerfile — Build multi-stage optimisé pour FastAPI
FROM python:3.11-slim AS builder

WORKDIR /app

# Copier requirements en premier pour bénéficier du cache Docker
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

# Image de production — plus légère
FROM python:3.11-slim AS production
WORKDIR /app

# Copier les packages installés depuis le builder
COPY --from=builder /root/.local /root/.local
COPY . .

# Ajouter les binaires locaux au PATH
ENV PATH=/root/.local/bin:$PATH
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

EXPOSE 8000

# Lancer Uvicorn en production (sans --reload)
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

Build Angular pour la production

# Build Angular optimisé pour la production
ng build --configuration=production

# Le dossier dist/my-app/ contient les fichiers statiques à servir
# Servir via Nginx, Apache ou un CDN (S3, Cloudflare Pages...)

Conclusion

Vous avez maintenant une application CRUD full-stack opérationnelle avec FastAPI, SQLAlchemy, Pydantic et Angular. Cette architecture vous offre :

• Performance : FastAPI est l'un des frameworks Python les plus rapides (Starlette + ASGI)
• Sécurité : Validation automatique des inputs, CORS strict, variables d'environnement
• Maintenabilité : Séparation claire entre modèles, schémas, CRUD et routes
• Testabilité : Tests d'intégration avec base de données en mémoire
• Documentation automatique : Swagger UI générée par FastAPI à /docs

Checklist de mise en production

Pour aller plus loin, consultez nos articles sur FastAPI : auth JWT et tests pour sécuriser vos endpoints, et explorez les patterns de cache Redis pour améliorer les performances de lecture.