Créer des API REST avec Python : guide intermédiaire

Ce tutoriel explique comment concevoir, implémenter, tester et déployer une API REST en Python avec un niveau intermédiaire. L’objectif est d’aller au-delà du « Hello World » : structure de projet, validation, gestion d’erreurs, authentification, pagination, versioning, tests, migrations, conteneurisation, et bonnes pratiques de production.

1) Prérequis et objectifs

Prérequis

• Python 3.11+ (3.10 minimum possible)
• Connaissances de base en HTTP (méthodes, codes de statut)
• Connaissances Python (modules, typage, exceptions)
• Un terminal (Linux/macOS) ou PowerShell (Windows)

Objectifs

À la fin, vous saurez :

• Créer une API REST structurée avec FastAPI
• Définir des schémas de données et valider les entrées
• Utiliser une base de données SQL (PostgreSQL) via SQLAlchemy
• Gérer les migrations avec Alembic
• Mettre en place une authentification par jeton (JWT)
• Ajouter pagination, filtrage, tri, versioning
• Tester l’API (unitaires + intégration) avec pytest
• Conteneuriser avec Docker et exécuter en production avec Uvicorn/Gunicorn

2) Choix technologiques (et pourquoi)

Il existe plusieurs frameworks Python pour les API : Flask, Django REST Framework, FastAPI, etc. Ici, on choisit FastAPI car :

• Validation et sérialisation via Pydantic
• Documentation interactive automatique (OpenAPI/Swagger)
• Excellentes performances (ASGI)
• Typage Python exploité à fond (autocomplétion, robustesse)

Pour la base de données :

• PostgreSQL : robuste, standard en production
• SQLAlchemy (2.x) : ORM mature, contrôle fin, requêtes expressives
• Alembic : migrations versionnées

3) Initialisation du projet

3.1 Créer un environnement virtuel

mkdir api-rest-python
cd api-rest-python
python -m venv .venv
source .venv/bin/activate

Sous Windows (PowerShell) :

python -m venv .venv
.\.venv\Scripts\Activate.ps1

3.2 Installer les dépendances

pip install fastapi uvicorn[standard] sqlalchemy psycopg[binary] pydantic-settings alembic python-jose[cryptography] passlib[bcrypt] pytest httpx

Optionnel (qualité de code) :

pip install ruff black

3.3 Arborescence recommandée

Une structure claire évite le « gros fichier unique » et facilite les tests.

api-rest-python/
  app/
    __init__.py
    main.py
    core/
      __init__.py
      config.py
      security.py
      database.py
    api/
      __init__.py
      v1/
        __init__.py
        router.py
        endpoints/
          __init__.py
          items.py
          auth.py
    models/
      __init__.py
      item.py
      user.py
    schemas/
      __init__.py
      item.py
      user.py
      token.py
    services/
      __init__.py
      items.py
      users.py
  tests/
    test_health.py
    test_items.py
  alembic.ini
  migrations/
  pyproject.toml (optionnel)
  Dockerfile (plus tard)
  docker-compose.yml (plus tard)

4) Configuration et paramètres (Pydantic Settings)

Créez app/core/config.py :

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    app_name: str = "API REST Python"
    environment: str = "dev"

    database_url: str = "postgresql+psycopg://postgres:postgres@localhost:5432/apidb"

    jwt_secret_key: str = "change-moi-en-production"
    jwt_algorithm: str = "HS256"
    jwt_access_token_exp_minutes: int = 30


settings = Settings()

Créez un fichier .env à la racine :

cat > .env << 'EOF'
ENVIRONMENT=dev
DATABASE_URL=postgresql+psycopg://postgres:postgres@localhost:5432/apidb
JWT_SECRET_KEY=super-secret-a-changer
JWT_ACCESS_TOKEN_EXP_MINUTES=30
EOF

Pourquoi c’est important :

• Les paramètres ne doivent pas être codés en dur.
• En production, vous injectez des variables d’environnement (secrets, URL DB, etc.).
• Vous pouvez avoir plusieurs environnements (dev, test, prod).

5) Base de données : moteur, session, dépendances FastAPI

Créez app/core/database.py :

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

from app.core.config import settings


class Base(DeclarativeBase):
    pass


engine = create_engine(settings.database_url, pool_pre_ping=True)

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)


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

Explications :

• pool_pre_ping=True évite des connexions mortes (utile en production).
• get_db() est une dépendance FastAPI : une session par requête, fermée proprement.

6) Modèles SQLAlchemy (User, Item)

6.1 Modèle User

Créez app/models/user.py :

from sqlalchemy import String, Integer, Boolean
from sqlalchemy.orm import Mapped, mapped_column

from app.core.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email: Mapped[str] = mapped_column(String(320), unique=True, index=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)

6.2 Modèle Item

Créez app/models/item.py :

from sqlalchemy import String, Integer, Text, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column

from app.core.database import Base


class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(120), index=True, nullable=False)
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    created_at: Mapped[object] = mapped_column(DateTime(timezone=True), server_default=func.now(), nullable=False)

Remarque : on utilise server_default=func.now() pour que la base gère la date de création.

7) Schémas Pydantic : validation et sérialisation

Les schémas décrivent ce que l’API accepte et renvoie. Ils servent à :

• Valider automatiquement les entrées
• Générer la documentation OpenAPI
• Éviter d’exposer des champs sensibles (ex. mot de passe)

7.1 Schémas User

Créez app/schemas/user.py :

from pydantic import BaseModel, EmailStr, Field


class UserCreate(BaseModel):
    email: EmailStr
    password: str = Field(min_length=8, max_length=128)


class UserPublic(BaseModel):
    id: int
    email: EmailStr
    is_active: bool

    class Config:
        from_attributes = True

7.2 Schémas Item

Créez app/schemas/item.py :

from pydantic import BaseModel, Field
from datetime import datetime


class ItemCreate(BaseModel):
    name: str = Field(min_length=1, max_length=120)
    description: str | None = Field(default=None, max_length=2000)


class ItemUpdate(BaseModel):
    name: str | None = Field(default=None, min_length=1, max_length=120)
    description: str | None = Field(default=None, max_length=2000)


class ItemPublic(BaseModel):
    id: int
    name: str
    description: str | None
    created_at: datetime

    class Config:
        from_attributes = True

8) Migrations avec Alembic

8.1 Initialiser Alembic

alembic init migrations

Dans alembic.ini, configurez l’URL (ou laissez vide et lisez depuis settings via env.py). Éditez migrations/env.py pour pointer vers vos métadonnées.

Ouvrez migrations/env.py et remplacez la partie configuration par :

from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

from app.core.config import settings
from app.core.database import Base
from app.models.user import User
from app.models.item import Item

config = context.config
fileConfig(config.config_file_name)

target_metadata = Base.metadata

def run_migrations_offline():
    context.configure(
        url=settings.database_url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online():
    connectable = engine_from_config(
        {"sqlalchemy.url": settings.database_url},
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(connection=connection, target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

8.2 Lancer PostgreSQL rapidement (Docker)

Si vous n’avez pas PostgreSQL localement :

docker run --name apidb -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=apidb -p 5432:5432 -d postgres:16

8.3 Créer et appliquer la migration

alembic revision --autogenerate -m "create users and items"
alembic upgrade head

9) Sécurité : hachage de mot de passe et JWT

Créez app/core/security.py :

from datetime import datetime, timedelta, timezone
from jose import jwt
from passlib.context import CryptContext

from app.core.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def hash_password(password: str) -> str:
    return pwd_context.hash(password)


def verify_password(password: str, hashed_password: str) -> bool:
    return pwd_context.verify(password, hashed_password)


def create_access_token(subject: str) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_access_token_exp_minutes)
    payload = {"sub": subject, "exp": expire}
    return jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)

Points essentiels :

• Ne stockez jamais un mot de passe en clair.
• JWT : sub identifie l’utilisateur (souvent email ou id).
• exp impose une expiration.

10) Couche services : logique métier isolée

Séparer la logique métier des endpoints rend le code testable et maintenable.

10.1 Service utilisateurs

Créez app/services/users.py :

from sqlalchemy.orm import Session
from sqlalchemy import select

from app.models.user import User
from app.core.security import hash_password, verify_password


def create_user(db: Session, email: str, password: str) -> User:
    user = User(email=email, hashed_password=hash_password(password))
    db.add(user)
    db.commit()
    db.refresh(user)
    return user


def get_user_by_email(db: Session, email: str) -> User | None:
    return db.scalar(select(User).where(User.email == email))


def authenticate_user(db: Session, email: str, password: str) -> User | None:
    user = get_user_by_email(db, email)
    if not user:
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

10.2 Service items

Créez app/services/items.py :

from sqlalchemy.orm import Session
from sqlalchemy import select, desc

from app.models.item import Item


def create_item(db: Session, name: str, description: str | None) -> Item:
    item = Item(name=name, description=description)
    db.add(item)
    db.commit()
    db.refresh(item)
    return item


def list_items(db: Session, limit: int, offset: int, q: str | None = None) -> list[Item]:
    stmt = select(Item)
    if q:
        stmt = stmt.where(Item.name.ilike(f"%{q}%"))
    stmt = stmt.order_by(desc(Item.id)).limit(limit).offset(offset)
    return list(db.scalars(stmt).all())


def get_item(db: Session, item_id: int) -> Item | None:
    return db.get(Item, item_id)


def update_item(db: Session, item: Item, name: str | None, description: str | None) -> Item:
    if name is not None:
        item.name = name
    if description is not None:
        item.description = description
    db.commit()
    db.refresh(item)
    return item


def delete_item(db: Session, item: Item) -> None:
    db.delete(item)
    db.commit()

11) API v1 : routeur, endpoints, gestion d’erreurs

11.1 Routeur principal v1

Créez app/api/v1/router.py :

from fastapi import APIRouter
from app.api.v1.endpoints import items, auth

router = APIRouter()
router.include_router(auth.router, prefix="/auth", tags=["auth"])
router.include_router(items.router, prefix="/items", tags=["items"])

11.2 Endpoint Auth (login + création utilisateur)

Créez app/schemas/token.py :

from pydantic import BaseModel


class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

Créez app/api/v1/endpoints/auth.py :

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.schemas.user import UserCreate, UserPublic
from app.schemas.token import Token
from app.services.users import create_user, get_user_by_email, authenticate_user
from app.core.security import create_access_token

router = APIRouter()


@router.post("/register", response_model=UserPublic, status_code=status.HTTP_201_CREATED)
def register(payload: UserCreate, db: Session = Depends(get_db)):
    existing = get_user_by_email(db, payload.email)
    if existing:
        raise HTTPException(status_code=409, detail="Email déjà utilisé")
    return create_user(db, payload.email, payload.password)


@router.post("/login", response_model=Token)
def login(payload: UserCreate, db: Session = Depends(get_db)):
    user = authenticate_user(db, payload.email, payload.password)
    if not user:
        raise HTTPException(status_code=401, detail="Identifiants invalides")
    token = create_access_token(subject=user.email)
    return Token(access_token=token)

Remarque : pour simplifier, on réutilise UserCreate pour le login (email + password). En production, vous pourriez créer un schéma LoginRequest.

11.3 Dépendance d’authentification (Bearer)

Créez app/api/v1/endpoints/items.py avec une protection simple. D’abord, créez une fonction utilitaire dans app/core/security.py (ajout) :

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError

security_scheme = HTTPBearer(auto_error=False)

def get_current_subject(credentials: HTTPAuthorizationCredentials | None = Depends(security_scheme)) -> str:
    if credentials is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Jeton manquant")
    token = credentials.credentials
    try:
        payload = jwt.decode(token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm])
        sub = payload.get("sub")
        if not sub:
            raise HTTPException(status_code=401, detail="Jeton invalide")
        return sub
    except JWTError:
        raise HTTPException(status_code=401, detail="Jeton invalide")

Ensuite, app/api/v1/endpoints/items.py :

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.core.security import get_current_subject
from app.schemas.item import ItemCreate, ItemPublic, ItemUpdate
from app.services.items import create_item, list_items, get_item, update_item, delete_item

router = APIRouter()


@router.get("/", response_model=list[ItemPublic])
def get_items(
    db: Session = Depends(get_db),
    limit: int = Query(20, ge=1, le=100),
    offset: int = Query(0, ge=0),
    q: str | None = Query(None, max_length=120),
):
    return list_items(db, limit=limit, offset=offset, q=q)


@router.post("/", response_model=ItemPublic, status_code=status.HTTP_201_CREATED)
def post_item(
    payload: ItemCreate,
    db: Session = Depends(get_db),
    subject: str = Depends(get_current_subject),
):
    # subject est l’email issu du JWT (utile pour audit/permissions)
    return create_item(db, payload.name, payload.description)


@router.get("/{item_id}", response_model=ItemPublic)
def get_one_item(item_id: int, db: Session = Depends(get_db)):
    item = get_item(db, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Ressource introuvable")
    return item


@router.patch("/{item_id}", response_model=ItemPublic)
def patch_item(
    item_id: int,
    payload: ItemUpdate,
    db: Session = Depends(get_db),
    subject: str = Depends(get_current_subject),
):
    item = get_item(db, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Ressource introuvable")
    return update_item(db, item, payload.name, payload.description)


@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def remove_item(
    item_id: int,
    db: Session = Depends(get_db),
    subject: str = Depends(get_current_subject),
):
    item = get_item(db, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Ressource introuvable")
    delete_item(db, item)
    return None

Explications :

• GET /items reste public, POST/PATCH/DELETE sont protégés par JWT.
• Pagination : limit et offset.
• Recherche simple : q filtre sur name via ilike.

12) Application FastAPI : point d’entrée, versioning, santé

Créez app/main.py :

from fastapi import FastAPI
from app.core.config import settings
from app.api.v1.router import router as v1_router

app = FastAPI(title=settings.app_name)

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

app.include_router(v1_router, prefix="/api/v1")

Versioning :

• Le préfixe /api/v1 permet d’introduire /api/v2 plus tard sans casser les clients.
• Évitez de versionner uniquement par en-tête si vous débutez : l’URL est plus simple à diagnostiquer.

13) Lancer le serveur et vérifier la documentation

Démarrage :

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

Ouvrez :

• Documentation Swagger : http://localhost:8000/docs
• OpenAPI JSON : http://localhost:8000/openapi.json
• Santé : http://localhost:8000/health

14) Tester l’API avec des commandes réelles (curl)

14.1 Santé

curl -s http://localhost:8000/health | python -m json.tool

14.2 Inscription

curl -s -X POST http://localhost:8000/api/v1/auth/register \
  -H "Content-Type: application/json" \
  -d '{"email":"alice@example.com","password":"motdepasse123"}' | python -m json.tool

14.3 Connexion (récupérer un jeton)

TOKEN=$(curl -s -X POST http://localhost:8000/api/v1/auth/login \
  -H "Content-Type: application/json" \
  -d '{"email":"alice@example.com","password":"motdepasse123"}' | python -c "import sys,json; print(json.load(sys.stdin)['access_token'])")

echo "$TOKEN"

14.4 Créer un item (protégé)

curl -s -X POST http://localhost:8000/api/v1/items \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $TOKEN" \
  -d '{"name":"Clavier","description":"Clavier mécanique"}' | python -m json.tool

14.5 Lister avec pagination et recherche

curl -s "http://localhost:8000/api/v1/items?limit=10&offset=0&q=clav" | python -m json.tool

15) Gestion d’erreurs : cohérence et bonnes pratiques

Dans les exemples, on lève HTTPException avec des codes :

• 401 : non authentifié / jeton invalide
• 404 : ressource absente
• 409 : conflit (email déjà utilisé)
• 422 : validation automatique FastAPI/Pydantic

Bonnes pratiques :

• Utiliser des messages d’erreur stables (utiles côté client).
• Éviter d’exposer des détails internes (stack traces, SQL brut).
• Documenter les erreurs attendues dans la description des endpoints (ou via OpenAPI avancé).

Vous pouvez aussi centraliser les erreurs via des gestionnaires d’exceptions (@app.exception_handler) pour uniformiser la réponse (ex. {"error": {"code": "...", "message": "..."}}).

16) Tests avec pytest et httpx

Les tests d’intégration vérifient que l’API répond correctement. Ici, on écrit des tests simples. Pour un vrai projet, vous isoleriez une base de test et des transactions.

16.1 Test de santé

Créez tests/test_health.py :

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_health():
    r = client.get("/health")
    assert r.status_code == 200
    assert r.json()["status"] == "ok"

16.2 Test items (exemple minimal)

Créez tests/test_items.py :

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_list_items():
    r = client.get("/api/v1/items?limit=5&offset=0")
    assert r.status_code == 200
    assert isinstance(r.json(), list)

Lancer :

pytest -q

Pour aller plus loin (recommandé) :

• Créer une base PostgreSQL dédiée aux tests
• Utiliser des fixtures pytest pour créer un utilisateur, obtenir un token, créer un item, puis nettoyer
• Remplacer get_db par une session de test (override FastAPI) afin de ne pas toucher la base de dev

17) Conteneurisation avec Docker (API + PostgreSQL)

17.1 Dockerfile

Créez Dockerfile :

FROM python:3.11-slim

WORKDIR /app

RUN pip install --no-cache-dir --upgrade pip

COPY . /app

RUN pip install --no-cache-dir fastapi uvicorn[standard] sqlalchemy psycopg[binary] pydantic-settings alembic python-jose[cryptography] passlib[bcrypt]

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

17.2 docker-compose.yml

Créez docker-compose.yml :

cat > docker-compose.yml << 'EOF'
services:
  db:
    image: postgres:16
    environment:
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: apidb
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data

  api:
    build: .
    environment:
      DATABASE_URL: postgresql+psycopg://postgres:postgres@db:5432/apidb
      JWT_SECRET_KEY: super-secret-a-changer
      JWT_ACCESS_TOKEN_EXP_MINUTES: 30
    ports:
      - "8000:8000"
    depends_on:
      - db

volumes:
  pgdata:
EOF

Démarrer :

docker compose up --build -d

Appliquer les migrations dans le conteneur API :

docker compose exec api alembic upgrade head

Vérifier :

curl -s http://localhost:8000/health

18) Production : Uvicorn/Gunicorn, logs, variables et sécurité

18.1 Serveur d’application

En production, on utilise souvent Gunicorn avec des workers Uvicorn :

pip install gunicorn
gunicorn -k uvicorn.workers.UvicornWorker -w 2 -b 0.0.0.0:8000 app.main:app

Ajustez -w selon CPU et charge.

18.2 Sécurité

• Changez JWT_SECRET_KEY (long, aléatoire, stocké en secret manager).
• Activez HTTPS (reverse proxy type Nginx/Traefik).
• Ajoutez des limites (rate limiting) et du monitoring.
• Mettez en place des permissions : aujourd’hui, n’importe quel utilisateur authentifié peut modifier n’importe quel item. En pratique, vous associez un owner_id à Item et vérifiez les droits.

18.3 Observabilité

• Logs structurés (JSON) si besoin.
• Traces et métriques (OpenTelemetry, Prometheus) selon contexte.

19) Améliorations intermédiaires recommandées

19.1 Pagination plus riche

Au lieu de limit/offset, vous pouvez renvoyer :

• total
• next_offset
• items

Exemple de réponse :

{
  "total": 120,
  "items": [...],
  "limit": 20,
  "offset": 0
}

Cela demande une requête COUNT(*) en plus, ou une stratégie plus avancée.

19.2 Tri et filtres

Ajoutez des paramètres sort et order, en validant strictement les champs autorisés pour éviter l’injection via noms de colonnes.

19.3 Versioning et dépréciation

• Conservez /api/v1 tant que des clients l’utilisent.
• Ajoutez /api/v2 avec les changements incompatibles.
• Documentez une politique de dépréciation (dates, en-têtes d’avertissement).

19.4 Validation métier

Pydantic valide la forme, mais pas toujours la logique métier :

• Interdire certains mots dans name
• Vérifier l’unicité d’un champ (nécessite DB)
• Appliquer des règles selon l’utilisateur (permissions)

20) Récapitulatif

Vous avez construit une API REST intermédiaire avec :

• FastAPI (routes, dépendances, docs)
• SQLAlchemy + PostgreSQL (persistance)
• Alembic (migrations)
• JWT (authentification)
• Pagination, recherche, gestion d’erreurs
• Tests (pytest)
• Docker (déploiement reproductible)

Prochaine étape naturelle : ajouter des relations (propriétaire d’item), des rôles (admin), une base de test isolée, et une stratégie de déploiement (CI/CD).

Annexes : commandes utiles

Arrêter et supprimer le conteneur PostgreSQL lancé via docker run :

docker stop apidb
docker rm apidb

Voir les logs :

docker compose logs -f api

Générer une nouvelle migration :

alembic revision --autogenerate -m "ma_nouvelle_migration"
alembic upgrade head

Formater et vérifier le code (optionnel) :

ruff check .
black .

Si vous souhaitez, je peux proposer une variante avec :

• Relations User ↔ Item (owner), permissions, et endpoints « mes items »
• Base de test PostgreSQL dédiée + overrides FastAPI propres
• Réponses d’erreurs uniformisées et schémas OpenAPI enrichis