MFormations
Modern Python Engineering

Chapitre 17

Chapitre 17 : Exercices Pratiques

> **40 exercices couvrant l'ensemble du programme** — de la syntaxe Python à l'architecture en production.

Chapitre 17 — Exercices Pratiques

Version : 1.0.0
Auteur : Modern Python Engineering Team
Prérequis : Python ≥ 3.11, notions des chapitres 1-16

Table des matières

  1. Introduction
  2. Module 1 : Python Basics (Exercices 01–08)
  3. Module 2 : Async (Exercices 09–14)
  4. Module 3 : pytest (Exercices 15–19)
  5. Module 4 : FastAPI (Exercices 20–24)
  6. Module 5 : SQLAlchemy (Exercices 25–28)
  7. Module 6 : pandas (Exercices 29–32)
  8. Module 7 : scikit-learn (Exercices 33–35)
  9. Module 8 : Architecture & CLI (Exercices 36–38)
  10. Module 9 : Docker (Exercices 39–40)
  11. Ressources complémentaires

Introduction

Ce chapitre contient 40 exercices progressifs qui couvrent l'intégralité du programme Modern Python Engineering. Chaque exercice est accompagné :

  • D'un énoncé détaillé avec contexte et attentes.
  • D'un jeu de tests pytest pour valider votre solution.
  • D'une correction commentée dans le chapitre 19.
  • D'une grille d'auto-évaluation.

Notation et conventions

# Les fonctions à implémenter sont annotées avec des types
def solve(data: list[int]) -> dict[str, float]: ...

# Les tests sont fournis dans tests/test_exNN.py
# Lancez-les avec :
# pytest tests/test_ex01.py -v

Structure d'un exercice

Chaque exercice se décompose en :

  1. Contexte : mise en situation réaliste (ex. "Vous devez parser un fichier CSV")
  2. Spécification : ce que doit faire la fonction/classe
  3. Contraintes : règles à respecter (pas de librairie externe, complexité O(n), etc.)
  4. Validation : commande pytest à exécuter
  5. Pour aller plus loin : variantes suggérées

Module 1 : Python Basics (Exercices 01–08)

Exercice 01 : Typage statique et validation

Contexte : Vous développez une librairie de validation de formulaires.

Spécification : Implémentez une fonction validate_email(email: str) -> bool qui vérifie qu'une adresse email respecte le format RFC 5322 simplifié.

def validate_email(email: str) -> bool:
    """Validate email format using RFC 5322 simplified rules.

    Args:
        email: The email string to validate.

    Returns:
        True if valid, False otherwise.
    """
    # À compléter

Contraintes :

  • N'utilisez pas re.match — implémentez l'algorithme manuellement.
  • Gérez les cas limites : "", "@", "a@b", " user@domain ".

Validation :

pytest tests/test_ex01.py -v

Pour aller plus loin : Ajoutez la validation des domaines avec vérification DNS.

Exercice 02 : Structures de données avancées

Contexte : Vous devez implémenter un cache LRU (Least Recently Used) sans utiliser functools.lru_cache.

Spécification :

class LRUCache[T: Hashable, U]:
    """LRU cache with O(1) get/put operations.

    Args:
        capacity: Maximum number of items.
    """

    def __init__(self, capacity: int) -> None: ...

    def get(self, key: T) -> U | None: ...

    def put(self, key: T, value: U) -> None: ...

Contraintes :

  • Complexité O(1) pour get et put.
  • Utilisez OrderedDict ou une dict doublée d'une list chaînée.
  • Le cache doit être thread-safe (utilisez threading.Lock).

Tests clés :

def test_lru_eviction():
    cache = LRUCache[str, int](2)
    cache.put("a", 1)
    cache.put("b", 2)
    cache.get("a")       # "a" devient récent
    cache.put("c", 3)    # "b" est évincé
    assert cache.get("b") is None
    assert cache.get("c") == 3

Exercice 03 : Itérateurs personnalisés

Contexte : Créez un itérateur qui génère la séquence de Fibonacci de manière paresseuse et potentiellement infinie.

class Fibonacci:
    """Iterator that yields Fibonacci numbers up to max_value."""

    def __init__(self, max_value: int | None = None) -> None: ...

    def __iter__(self) -> Self: ...

    def __next__(self) -> int: ...

Pour aller plus loin : Implémentez __getitem__ pour permettre l'indexation fib[10] et le slicing fib[5:15].

Exercice 04 : Générateurs et expressions génératrices

Contexte : Traitez un flux de logs (fichier de plusieurs Go) sans le charger en mémoire.

def parse_log_stream(
    filepath: Path,
    level: str = "ERROR",
) -> Generator[dict[str, Any], None, None]:
    """Yield parsed log entries of the specified level.

    Format de log attendu :
    [TIMESTAMP] LEVEL module: message
    """

Contraintes :

  • Utilisez yield pour produire les entrées une par une.
  • Ne chargez jamais plus d'une ligne en mémoire.

Exercice 05 : Décorateurs paramétrés

Contexte : Implémentez un décorateur de retry avec backoff exponentiel.

def retry(
    max_attempts: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    exceptions: tuple[type[Exception], ...] = (Exception,),
) -> Callable[..., Callable[..., Any]]:
    """Decorator that retries a function on failure.

    Args:
        max_attempts: Maximum number of attempts.
        delay: Initial delay in seconds.
        backoff: Multiplier applied to delay after each attempt.
        exceptions: Tuple of exception types to catch.
    """

Exemple d'utilisation :

@retry(max_attempts=5, delay=0.1, exceptions=(ConnectionError, TimeoutError))
def fetch_data(url: str) -> bytes: ...

Exercice 06 : Context managers

Contexte : Implémentez un context manager qui mesure le temps d'exécution d'un bloc de code, avec un rapport optionnel.

@contextmanager
def timed(label: str = "", report: bool = True) -> Generator[float, None, None]:
    """Context manager that times code execution.

    Yields:
        The elapsed time in seconds when exiting the context.
    """

Exercice 07 : Dataclasses et validation

Contexte : Modélisez un système de réservation de vols avec validation.

@dataclass(frozen=True, slots=True)
class Booking:
    passenger_name: str
    flight_number: str
    seat: str | None = None
    extra_baggage: bool = False

Validation :

  • passenger_name : 2-100 caractères, lettres uniquement
  • flight_number : pattern XX1234 (2 lettres + 4 chiffres)
  • seat : optionnel, pattern 12A (nombre + lettre)

Utilisez __post_init__ pour les validations.

Exercice 08 : Programmation fonctionnelle

Contexte : Traitez une liste de transactions bancaires avec map, filter, reduce et itertools.

Transaction = namedtuple("Transaction", ["date", "amount", "category"])

def summarize_by_category(
    transactions: list[Transaction],
) -> dict[str, float]:
    """Summarize transactions by category using functional programming."""

Contraintes : Pas de boucles for ou while. Utilisez exclusivement map, filter, functools.reduce et itertools.groupby.

Module 2 : Async (Exercices 09–14)

Exercice 09 : Coroutines de base

Contexte : Simulez des appels API asynchrones avec des délais variables.

async def fetch_url(url: str, delay: float = 1.0) -> dict[str, Any]:
    """Simulate fetching a URL with a given delay."""
    await asyncio.sleep(delay)
    return {"url": url, "status": 200, "data": f"Content of {url}"}

async def fetch_all(urls: list[str]) -> list[dict[str, Any]]:
    """Fetch all URLs concurrently using asyncio.gather."""

Exercice 10 : asyncio.gather et gestion d'erreurs

Contexte : Implémentez un gather avec gestion fine des erreurs : au lieu d'annuler toutes les tâches à la première erreur, collectez les résultats partiels.

async def gather_with_errors(
    *coros: Awaitable[Any],
    return_exceptions: bool = True,
) -> list[Any]:
    """Gather results, collecting exceptions without cancelling others."""

Exercice 11 : Async Queue et producteur-consommateur

Contexte : Implémentez un pipeline de traitement d'images avec une file d'attente asynchrone.

async def producer(queue: asyncio.Queue[ImageTask], images: list[Path]) -> None: ...
async def worker(
    queue: asyncio.Queue[ImageTask],
    result_queue: asyncio.Queue[Result],
    worker_id: int,
) -> None: ...
async def consumer(result_queue: asyncio.Queue[Result]) -> None: ...

Exercice 12 : Timeout et cancellation

Contexte : Implémentez une fonction avec timeout et fallback.

async def fetch_with_timeout(
    url: str,
    timeout: float = 5.0,
    fallback: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Fetch URL with timeout. Return fallback if timeout occurs."""

Utilisez asyncio.timeout (Python 3.12+) ou asyncio.wait_for.

Exercice 13 : aiofiles et I/O asynchrone

Contexte : Traitez un répertoire de fichiers CSV asynchrone.

async def process_csv_files(
    directory: Path,
    chunk_size: int = 8192,
) -> AsyncIterator[dict[str, str]]:
    """Read and parse CSV files asynchronously using aiofiles."""

Exercice 14 : AnyIO — library-agnostic async

Contexte : Réécrivez les exercices 09-13 en utilisant AnyIO pour être compatible avec asyncio et trio.

import anyio

async def run_pipeline() -> None:
    async with anyio.create_task_group() as tg:
        tg.start_soon(fetch_url, "https://example.com")

Module 3 : pytest (Exercices 15–19)

Exercice 15 : Fixtures et scopes

Contexte : Créez une fixture pour une base de données SQLite en mémoire.

@pytest.fixture
def db_connection():
    """Provide a fresh SQLite in-memory database for each test."""
    conn = sqlite3.connect(":memory:")
    conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
    yield conn
    conn.close()

Niveaux :

  • @pytest.fixture simple
  • scope="module"
  • scope="session"
  • autouse=True

Exercice 16 : Mocking avec unittest.mock

Contexte : Testez une fonction qui appelle une API externe.

def get_weather(city: str) -> dict[str, Any]:
    response = requests.get(f"https://api.weather.com/v1/{city}")
    response.raise_for_status()
    return response.json()

# À tester sans appeler l'API réelle
def test_get_weather(httpx_mock: HTTPXMock) -> None: ...

Exercice 17 : Paramétrage avancé

Contexte : Testez une fonction avec 20 combinaisons d'entrées.

@pytest.mark.parametrize(
    "input_data,expected",
    [
        ([1, 2, 3], 6),
        ([], 0),
        ([-1, 0, 1], 0),
        pytest.param([1], 1, id="single_element"),
        pytest.param(
            [10**6] * 1000, 10**9,
            marks=pytest.mark.slow,
            id="large_input",
        ),
    ],
)
def test_sum_list(input_data: list[int], expected: int) -> None:
    assert sum_list(input_data) == expected

Exercice 18 : Tests de propriétés avec Hypothesis

Contexte : Vérifiez que votre implémentation de LRU cache est correcte via des tests basés sur les propriétés.

from hypothesis import given, strategies as st

@given(st.lists(st.integers()), st.integers(min_value=1, max_value=100))
def test_lru_cache_properties(items: list[int], capacity: int) -> None:
    cache = LRUCache[int, int](capacity)
    # Vérifier que get(put(x)) == x
    for x in items:
        cache.put(x, x)
        assert cache.get(x) == x

Exercice 19 : Tests de non-régression et coverage

Contexte : Établissez une suite de tests de non-régression pour un module de calcul financier avec un coverage ≥ 95 %.

def test_coverage_threshold() -> None:
    """Vérifier que le coverage est ≥ 95 %."""
    # Utiliser pytest-cov --fail-under=95
    pass

Module 4 : FastAPI (Exercices 20–24)

Exercice 20 : CRUD REST de base

Contexte : Implémentez une API de gestion de tâches (Todo list).

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class TodoCreate(BaseModel):
    title: str
    description: str = ""

class Todo(TodoCreate):
    id: int
    completed: bool = False

@app.post("/todos", response_model=Todo, status_code=201)
async def create_todo(todo: TodoCreate) -> Todo: ...
@app.get("/todos")
async def list_todos() -> list[Todo]: ...
@app.get("/todos/{todo_id}")
async def get_todo(todo_id: int) -> Todo: ...
@app.put("/todos/{todo_id}")
async def update_todo(todo_id: int, todo: TodoCreate) -> Todo: ...
@app.delete("/todos/{todo_id}", status_code=204)
async def delete_todo(todo_id: int) -> None: ...

Exercice 21 : Validation avancée et dépendances

Contexte : Ajoutez validation, pagination, tri et dépendances.

from fastapi import Depends, Query

async def pagination_params(
    page: int = Query(1, ge=1),
    size: int = Query(20, ge=1, le=100),
) -> tuple[int, int]: ...

@app.get("/todos")
async def list_todos(
    pagination: tuple[int, int] = Depends(pagination_params),
    completed: bool | None = None,
) -> list[Todo]: ...

Exercice 22 : Authentification et sécurité

Contexte : Implémentez JWT authentication avec OAuth2 Password Flow.

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    payload = decode_jwt(token)
    return await get_user(payload["sub"])

Exercice 23 : WebSocket temps réel

Contexte : Implémentez un chat temps réel avec WebSocket.

@app.websocket("/ws/chat/{room_id}")
async def chat_websocket(
    websocket: WebSocket,
    room_id: str,
) -> None:
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")

Exercice 24 : Tests d'API avec TestClient

Contexte : Écrivez des tests complets pour l'API Todo.

from fastapi.testclient import TestClient

client = TestClient(app)

def test_create_todo():
    response = client.post("/todos", json={"title": "Learn FastAPI"})
    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Learn FastAPI"
    assert "id" in data

Module 5 : SQLAlchemy (Exercices 25–28)

Exercice 25 : ORM — Modèles et relations

Contexte : Modélisez un blog avec utilisateurs, articles et commentaires.

from sqlalchemy.orm import Mapped, mapped_column, relationship

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

Exercice 26 : Core — Requêtes SQL sans ORM

Contexte : Exécutez des requêtes avancées avec SQLAlchemy Core.

from sqlalchemy import text, select, func

async def get_top_authors(session: AsyncSession, limit: int = 10) -> list[dict]:
    stmt = (
        select(User.name, func.count(Post.id).label("post_count"))
        .join(Post, User.id == Post.author_id)
        .group_by(User.id)
        .order_by(func.count(Post.id).desc())
        .limit(limit)
    )
    result = await session.execute(stmt)
    return result.mappings().all()

Exercice 27 : Alembic — Migrations

Contexte : Créez et gérez des migrations de schéma.

alembic init alembic
alembic revision --autogenerate -m "add users table"
alembic upgrade head

Exercice 28 : Async Session et unit of work

Contexte : Implémentez un pattern Unit of Work avec session async.

class UnitOfWork:
    def __init__(self, session_factory: AsyncSessionFactory) -> None:
        self.session_factory = session_factory

    async def __aenter__(self) -> Self:
        self.session: AsyncSession = self.session_factory()
        return self

    async def __aexit__(self, *args: Any) -> None:
        await self.session.close()

Module 6 : pandas (Exercices 29–32)

Exercice 29 : DataFrames et nettoyage

Contexte : Nettoyez un dataset de ventes avec valeurs manquantes et outliers.

def clean_sales_data(df: pd.DataFrame) -> pd.DataFrame:
    """Clean sales DataFrame:
    - Remove rows with missing critical fields
    - Cap outliers at 99th percentile
    - Standardize date formats
    - Remove duplicates
    """

Exercice 30 : GroupBy et agrégations

Contexte : Analysez les performances commerciales par région et catégorie.

def monthly_summary(df: pd.DataFrame) -> pd.DataFrame:
    return (
        df.groupby([pd.Grouper(key="date", freq="ME"), "region", "category"])
        .agg(total_sales=("amount", "sum"), count=("id", "nunique"))
        .reset_index()
    )

Exercice 31 : Merge et Jointures

Contexte : Combinez 3 DataFrames (commandes, clients, produits) en une vue unique.

def build_customer_view(
    orders: pd.DataFrame,
    customers: pd.DataFrame,
    products: pd.DataFrame,
) -> pd.DataFrame:
    return (
        orders
        .merge(customers, on="customer_id", how="left")
        .merge(products, on="product_id", how="left")
    )

Exercice 32 : Time Series

Contexte : Analysez une série temporelle de prix d'actions avec rééchantillonnage et fenêtres glissantes.

def resample_ohlc(df: pd.DataFrame, freq: str = "1D") -> pd.DataFrame:
    return df.resample(freq).agg({
        "price": ["first", "max", "min", "last"],
        "volume": "sum",
    })

Module 7 : scikit-learn (Exercices 33–35)

Exercice 33 : Pipeline de classification

Contexte : Construisez un pipeline complet de classification pour le dataset Iris.

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier

pipeline = Pipeline([
    ("scaler", StandardScaler()),
    ("classifier", RandomForestClassifier(
        n_estimators=100, random_state=42
    )),
])

Exercice 34 : Grid Search et validation croisée

Contexte : Optimisez les hyperparamètres d'un modèle XGBoost.

from sklearn.model_selection import GridSearchCV

param_grid = {
    "classifier__n_estimators": [100, 200],
    "classifier__max_depth": [3, 5, 7],
    "classifier__learning_rate": [0.01, 0.1],
}

grid = GridSearchCV(pipeline, param_grid, cv=5, scoring="f1")
grid.fit(X_train, y_train)

Exercice 35 : Feature engineering et sélection

Contexte : Créez un pipeline de feature engineering pour données tabulaires.

class FeatureEngineer(TransformerMixin, BaseEstimator):
    def __init__(self, add_polynomials: bool = True) -> None:
        self.add_polynomials = add_polynomials

    def fit(self, X: np.ndarray, y: np.ndarray | None = None) -> Self: ...
    def transform(self, X: np.ndarray) -> np.ndarray: ...

Module 8 : Architecture & CLI (Exercices 36–38)

Exercice 36 : Clean Architecture

Contexte : Organisez un projet selon les principes de clean architecture.

src/
├── domain/          # Entities, value objects
│   ├── entities.py
│   └── value_objects.py
├── application/     # Use cases, ports
│   ├── use_cases.py
│   └── ports.py
├── infrastructure/  # Adapters (DB, API, etc.)
│   ├── database.py
│   └── repositories.py
└── interfaces/       # Entry points (API, CLI)
    ├── api/
    └── cli/

Exercice 37 : CLI avec Typer

Contexte : Créez un outil CLI pour analyser des fichiers de log.

import typer
from rich.console import Console
from rich.table import Table

app = typer.Typer()

@app.command()
def analyze(
    file: Path = typer.Argument(..., help="Path to log file"),
    level: str = typer.Option("ERROR", "--level", "-l"),
    output: Path | None = typer.Option(None, "--output", "-o"),
) -> None: ...

Exercice 38 : Injection de dépendances

Contexte : Implémentez un conteneur DI et l'appliquez à l'architecture clean.

from dependency_injector import containers, providers

class Container(containers.DeclarativeContainer):
    config = providers.Configuration()
    database = providers.Singleton(Database, url=config.db.url)
    repository = providers.Factory(UserRepository, session=database.provided.session)
    use_case = providers.Factory(CreateUserUseCase, repo=repository)

Module 9 : Docker (Exercices 39–40)

Exercice 39 : Dockerfile multi-stage

Contexte : Créez un Dockerfile multi-stage pour l'API FastAPI du module 4.

FROM python:3.13-slim AS builder
# Build dependencies

FROM python:3.13-slim AS production
# Runtime only

FROM production AS development
# Dev tooling + hot reload

Exercice 40 : docker-compose avec dépendances

Contexte : Créez un docker-compose avec FastAPI, PostgreSQL, Redis, et Celery worker.

version: "3.9"
services:
  api:
    build:
      target: development
    ports: ["8000:8000"]
    depends_on: [db, redis]

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: app
      POSTGRES_USER: app
      POSTGRES_PASSWORD: secret

  redis:
    image: redis:7-alpine

  worker:
    build:
      target: development
    command: celery -A src.worker worker -l info
    depends_on: [redis]

Ressources complémentaires

Prochain chapitre : 18-QCM →
Évaluez vos connaissances avec 100 questions à choix multiples couvrant l'ensemble du programme.

PrécédentChapitre 16 — 16 - Projet Fil Rouge : DataPipelineSuivantChapitre 18 — Chapitre 18 : QCM — 100 Questions à Choix Multiples