Chapitre 19 — Corrections Détaillées

Version : 1.0.0
Ce chapitre contient les corrigés complets des 40 exercices et des 100 QCM.

---

Table des matières

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

---

Exercices 01–08 : Python Basics

Correction — Exercice 01 : Validation d'email

Analyse : Un email RFC 5322 simplifié a la forme local-part@domain. Règles : local-part contient des lettres, chiffres et quelques caractères spéciaux (. _ % + -), domain contient des labels séparés par des points.

def validate_email(email: str) -> bool:
    """Validate email format using RFC 5322 simplified rules."""
    if not isinstance(email, str):
        return False

    email = email.strip()
    if not email:
        return False

    parts = email.split("@")
    if len(parts) != 2:
        return False

    local_part, domain = parts

    if not local_part or not domain:
        return False

    if len(local_part) > 64:
        return False

    allowed_local = set("abcdefghijklmnopqrstuvwxyz"
                        "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
                        "0123456789._%+-")

    for char in local_part:
        if char not in allowed_local:
            return False

    if local_part.startswith(".") or local_part.endswith("."):
        return False

    if ".." in local_part:
        return False

    domain_labels = domain.split(".")
    if len(domain_labels) < 2:
        return False

    allowed_domain = set("abcdefghijklmnopqrstuvwxyz"
                         "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
                         "0123456789-")

    for label in domain_labels:
        if not label:
            return False
        if len(label) > 63:
            return False
        if label.startswith("-") or label.endswith("-"):
            return False
        for char in label:
            if char not in allowed_domain:
                return False

    return True

Tests :

def test_validate_email():
    assert validate_email("user@example.com") is True
    assert validate_email("user.name+tag@example.co.uk") is True
    assert validate_email("") is False
    assert validate_email("notanemail") is False
    assert validate_email("@domain.com") is False
    assert validate_email("user@") is False
    assert validate_email("user@.com") is False
    assert validate_email("user@domain") is False
    assert validate_email("a@b.c") is True
    assert validate_email(" user@domain.com ") is True

Pièges :

• Oublier de strip() l'entrée
• Ne pas vérifier la longueur max (64 pour local-part, 254 total)
• Autoriser les points consécutifs dans le domaine

---

Correction — Exercice 02 : Cache LRU

from collections import OrderedDict
import threading

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

    def __init__(self, capacity: int) -> None:
        if capacity <= 0:
            msg = "Capacity must be positive"
            raise ValueError(msg)
        self.capacity = capacity
        self._cache: OrderedDict[T, U] = OrderedDict()
        self._lock = threading.Lock()

    def get(self, key: T) -> U | None:
        with self._lock:
            if key not in self._cache:
                return None
            self._cache.move_to_end(key)
            return self._cache[key]

    def put(self, key: T, value: U) -> None:
        with self._lock:
            self._cache[key] = value
            self._cache.move_to_end(key)
            if len(self._cache) > self.capacity:
                self._cache.popitem(last=False)

    def __len__(self) -> int:
        return len(self._cache)

    def __contains__(self, key: T) -> bool:
        return key in self._cache

Variante : Version sans OrderedDict utilisant une dict standard (Python 3.7+) et une list chaînée maison pour l'ordre.

---

Correction — Exercice 03 : Itérateur Fibonacci

from collections.abc import Iterator
from typing import Self

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

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

    def __iter__(self) -> Self:
        self.a = 0
        self.b = 1
        return self

    def __next__(self) -> int:
        result = self.a
        if self.max_value is not None and result > self.max_value:
            raise StopIteration
        self.a, self.b = self.b, self.a + self.b
        return result

    def __getitem__(self, index: int | slice) -> int | list[int]:
        if isinstance(index, slice):
            return list(self)[index]
        for i, val in enumerate(self):
            if i == index:
                return val
        raise IndexError("Fibonacci index out of range")

---

Correction — Exercice 04 : Parseur de logs

from collections.abc import Generator
from datetime import datetime
from pathlib import Path
import re

LOG_PATTERN = re.compile(
    r"\[(?P<timestamp>[^\]]+)\]\s+"
    r"(?P<level>DEBUG|INFO|WARNING|ERROR|CRITICAL)\s+"
    r"(?P<module>\w+):\s+"
    r"(?P<message>.+)"
)

def parse_log_stream(
    filepath: Path,
    level: str = "ERROR",
) -> Generator[dict[str, object], None, None]:
    """Yield parsed log entries of the specified level."""
    with open(filepath, encoding="utf-8") as f:
        for line in f:
            match = LOG_PATTERN.match(line.strip())
            if not match:
                continue
            entry = match.groupdict()
            if entry["level"] == level:
                entry["timestamp"] = datetime.fromisoformat(entry["timestamp"])
                yield entry

---

Correction — Exercice 05 : Décorateur retry

import time
import functools
from collections.abc import Callable
from typing import Any, TypeVar

F = TypeVar("F", bound=Callable[..., Any])

def retry(
    max_attempts: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    exceptions: tuple[type[Exception], ...] = (Exception,),
) -> Callable[[F], F]:
    """Decorator that retries a function on failure."""
    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            last_exception: Exception | None = None
            current_delay = delay
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt == max_attempts - 1:
                        raise
                    time.sleep(current_delay)
                    current_delay *= backoff
            raise RuntimeError("Unreachable")
        return wrapper  # type: ignore[return-value]
    return decorator

---

Correction — Exercice 06 : Context manager chronomètre

from collections.abc import Generator
from contextlib import contextmanager
import time

@contextmanager
def timed(label: str = "", report: bool = True) -> Generator[float, None, None]:
    """Context manager that times code execution."""
    start = time.perf_counter()
    try:
        yield 0.0
    finally:
        elapsed = time.perf_counter() - start
        if report:
            msg = f"{label}: {elapsed:.4f}s" if label else f"{elapsed:.4f}s"
            print(msg)
        yield elapsed  # type: ignore[misc]  # contextmanager ne yield qu'une fois

Note : Pour un vrai context manager avec yield final, utilisez une classe :

class timed:
    def __init__(self, label: str = "", report: bool = True) -> None:
        self.label = label
        self.report = report

    def __enter__(self) -> float:
        self.start = time.perf_counter()
        return 0.0

    def __exit__(self, *args: Any) -> None:
        self.elapsed = time.perf_counter() - self.start
        if self.report:
            msg = f"{self.label}: {self.elapsed:.4f}s" if self.label else f"{self.elapsed:.4f}s"
            print(msg)

---

Correction — Exercice 07 : Dataclasses avec validation

from dataclasses import dataclass
import re

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

    def __post_init__(self) -> None:
        if not (2 <= len(self.passenger_name) <= 100):
            raise ValueError("passenger_name must be 2-100 characters")
        if not self.passenger_name.replace(" ", "").isalpha():
            raise ValueError("passenger_name must contain only letters")

        if not re.fullmatch(r"[A-Z]{2}\d{4}", self.flight_number):
            raise ValueError("flight_number must match XX1234 pattern")

        if self.seat is not None:
            if not re.fullmatch(r"\d{1,2}[A-F]", self.seat):
                raise ValueError("seat must match pattern like 12A")

---

Correction — Exercice 08 : Programmation fonctionnelle

from collections import namedtuple
from functools import reduce
from itertools import groupby
from operator import itemgetter

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

def summarize_by_category(transactions: list[Transaction]) -> dict[str, float]:
    """Summarize transactions by category using functional programming."""
    sorted_tx = sorted(transactions, key=itemgetter(2))
    grouped = groupby(sorted_tx, key=itemgetter(2))

    def accumulate(acc: dict[str, float], item: tuple[str, object]) -> dict[str, float]:
        category, group = item
        total = reduce(lambda s, t: s + t.amount, group, 0.0)
        acc[category] = round(total, 2)
        return acc

    return reduce(accumulate, grouped, {})

---

Exercices 09–14 : Async

Correction — Exercice 09 : Coroutines de base

import asyncio

async def fetch_url(url: str, delay: float = 1.0) -> dict[str, object]:
    """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, object]]:
    """Fetch all URLs concurrently."""
    tasks = [fetch_url(url) for url in urls]
    results = await asyncio.gather(*tasks)
    return list(results)

---

Correction — Exercice 10 : gather avec gestion d'erreurs

import asyncio

async def gather_with_errors(
    *coros: object,
    return_exceptions: bool = True,
) -> list[object]:
    """Gather results, collecting exceptions without cancelling others."""
    tasks = [asyncio.ensure_future(c) for c in coros]  # type: ignore[arg-type]
    results: list[object] = []

    for task in tasks:
        try:
            result = await task
            results.append(result)
        except Exception as e:
            if return_exceptions:
                results.append(e)
            else:
                raise

    return results

---

Correction — Exercice 11 : Producteur-consommateur

import asyncio
from dataclasses import dataclass, field

@dataclass
class ImageTask:
    path: str
    size: tuple[int, int] = (0, 0)

@dataclass
class Result:
    task_id: str
    success: bool
    data: bytes | None = None

async def producer(
    queue: asyncio.Queue[ImageTask],
    images: list[str],
) -> None:
    for path in images:
        await queue.put(ImageTask(path=path))
        print(f"Produced: {path}")

async def worker(
    queue: asyncio.Queue[ImageTask],
    result_queue: asyncio.Queue[Result],
    worker_id: int,
) -> None:
    while True:
        task = await queue.get()
        try:
            await asyncio.sleep(0.1)
            result = Result(task_id=task.path, success=True, data=b"processed")
            print(f"Worker {worker_id} processed: {task.path}")
        except Exception as e:
            result = Result(task_id=task.path, success=False)
        await result_queue.put(result)
        queue.task_done()

async def consumer(result_queue: asyncio.Queue[Result]) -> None:
    while True:
        result = await result_queue.get()
        print(f"Consumed: {result.task_id} success={result.success}")
        result_queue.task_done()

---

Correction — Exercice 12 : Timeout et fallback

import asyncio

async def fetch_with_timeout(
    url: str,
    timeout: float = 5.0,
    fallback: dict[str, object] | None = None,
) -> dict[str, object]:
    """Fetch URL with timeout. Return fallback if timeout occurs."""
    try:
        async with asyncio.timeout(timeout):
            return await fetch_url(url, delay=2.0)
    except TimeoutError:
        return fallback or {"url": url, "error": "timeout"}

---

Correction — Exercice 13 : aiofiles

import aiofiles
from collections.abc import AsyncIterator
from pathlib import Path
import csv
import io

async def process_csv_files(
    directory: Path,
    chunk_size: int = 8192,
) -> AsyncIterator[dict[str, str]]:
    """Read and parse CSV files asynchronously."""
    for csv_file in directory.glob("*.csv"):
        async with aiofiles.open(csv_file, mode="r", encoding="utf-8") as f:
            content = await f.read()
            reader = csv.DictReader(io.StringIO(content))
            for row in reader:
                yield dict(row)

---

Correction — Exercice 14 : AnyIO

import anyio

async def run_pipeline() -> None:
    async with anyio.create_task_group() as tg:
        tg.start_soon(anyio.sleep, 1)
        tg.start_soon(anyio.sleep, 2)

anyio.run(run_pipeline)

---

Exercices 15–19 : pytest

Correction — Exercice 15 : Fixtures et scopes

import pytest
import sqlite3

@pytest.fixture
def db_connection():
    conn = sqlite3.connect(":memory:")
    conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
    conn.execute("INSERT INTO users VALUES (1, 'Alice')")
    yield conn
    conn.close()

def test_get_user(db_connection):
    cursor = db_connection.execute("SELECT name FROM users WHERE id=1")
    assert cursor.fetchone()[0] == "Alice"

@pytest.fixture(scope="session")
def config():
    return {"database": ":memory:", "debug": False}

---

Correction — Exercice 16 : Mocking

import pytest
import httpx

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

def test_get_weather(httpx_mock):
    httpx_mock.add_response(
        url="https://api.weather.com/v1/Paris",
        json={"city": "Paris", "temp": 22},
    )
    result = get_weather("Paris")
    assert result == {"city": "Paris", "temp": 22}

---

Correction — Exercices 17-19 : Paramétrage, Hypothesis, Coverage

import pytest
from hypothesis import given, strategies as st
from lru_cache import LRUCache

@pytest.mark.parametrize("input_data,expected", [
    ([1, 2, 3], 6),
    ([], 0),
    ([-1, 0, 1], 0),
], ids=["normal", "empty", "mixed"])
def test_sum_list(input_data, expected):
    assert sum(input_data) == expected

@given(st.lists(st.integers()), st.integers(min_value=1, max_value=100))
def test_lru_cache_properties(items, capacity):
    cache = LRUCache[int, int](capacity)
    for x in items:
        cache.put(x, x)
    for x in items:
        val = cache.get(x)
        assert val is None or val == x

def test_coverage_95():
    import subprocess, json
    result = subprocess.run(
        ["pytest", "--cov=src", "--cov-report=json", "-q"],
        capture_output=True, text=True,
    )
    with open("coverage.json") as f:
        data = json.load(f)
    assert data["totals"]["percent_covered"] >= 95

---

Exercices 20–24 : FastAPI

Correction — Exercice 20 : CRUD Todo

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uuid

app = FastAPI(title="Todo API")

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

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

todos: dict[str, Todo] = {}

@app.post("/todos", response_model=Todo, status_code=201)
async def create_todo(todo: TodoCreate) -> Todo:
    new = Todo(id=str(uuid.uuid4()), **todo.model_dump())
    todos[new.id] = new
    return new

@app.get("/todos")
async def list_todos() -> list[Todo]:
    return list(todos.values())

@app.get("/todos/{todo_id}")
async def get_todo(todo_id: str) -> Todo:
    if todo_id not in todos:
        raise HTTPException(404, "Todo not found")
    return todos[todo_id]

@app.put("/todos/{todo_id}")
async def update_todo(todo_id: str, todo: TodoCreate) -> Todo:
    if todo_id not in todos:
        raise HTTPException(404, "Todo not found")
    updated = Todo(id=todo_id, **todo.model_dump())
    todos[todo_id] = updated
    return updated

@app.delete("/todos/{todo_id}", status_code=204)
async def delete_todo(todo_id: str) -> None:
    if todo_id not in todos:
        raise HTTPException(404, "Todo not found")
    del todos[todo_id]

---

Correction — Exercice 24 : Tests API

from fastapi.testclient import TestClient
from main import app

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
    assert data["completed"] is False

def test_list_todos():
    client.post("/todos", json={"title": "Task 1"})
    response = client.get("/todos")
    assert response.status_code == 200
    assert len(response.json()) >= 1

def test_get_nonexistent():
    response = client.get("/todos/nonexistent")
    assert response.status_code == 404

def test_delete_todo():
    create_resp = client.post("/todos", json={"title": "To delete"})
    todo_id = create_resp.json()["id"]
    del_resp = client.delete(f"/todos/{todo_id}")
    assert del_resp.status_code == 204
    get_resp = client.get(f"/todos/{todo_id}")
    assert get_resp.status_code == 404

---

Exercices 25–28 : SQLAlchemy

Correction — Exercice 25 : ORM

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import ForeignKey, String, Text
from datetime import datetime

class Base(DeclarativeBase):
    pass

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

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    body: Mapped[str] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    author: Mapped["User"] = relationship(back_populates="posts")

---

Corrections exercices 26–28 (SQLAlchemy Core, Alembic, UoW)

# Exercice 26: Core
from sqlalchemy import select, func

async def get_top_authors(session, limit=10):
    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 28: Unit of Work
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

class UnitOfWork:
    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
        self.session_factory = session_factory

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

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

---

Exercices 29–32 : pandas

Correction — Exercice 29 : Clean sales data

import pandas as pd
import numpy as np

def clean_sales_data(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    critical_cols = ["date", "amount", "product"]
    df = df.dropna(subset=critical_cols)
    df = df.drop_duplicates()
    df["date"] = pd.to_datetime(df["date"], errors="coerce")
    df = df.dropna(subset=["date"])
    cap = df["amount"].quantile(0.99)
    df["amount"] = df["amount"].clip(upper=cap)
    return df

---

Corrections exercices 30–32 (GroupBy, Merge, Time Series)

# Exercice 30
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
def build_customer_view(orders, customers, products):
    return (
        orders
        .merge(customers, on="customer_id", how="left")
        .merge(products, on="product_id", how="left")
    )

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

---

Exercices 33–35 : scikit-learn

Correction — Exercice 33 : Pipeline classification

from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report

iris = load_iris()
X_train, X_test, y_train, y_test = train_test_split(
    iris.data, iris.target, test_size=0.2, random_state=42
)

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

pipeline.fit(X_train, y_train)
y_pred = pipeline.predict(X_test)
print(classification_report(y_test, y_pred, target_names=iris.target_names))

---

Exercices 34–35 : Grid Search et Feature Engineering

# Exercice 34
from sklearn.model_selection import GridSearchCV

param_grid = {
    "classifier__n_estimators": [100, 200],
    "classifier__max_depth": [3, 5, 7],
}
grid = GridSearchCV(pipeline, param_grid, cv=5, scoring="f1_macro")
grid.fit(X_train, y_train)
print(f"Best params: {grid.best_params_}, score: {grid.best_score_:.3f}")

# Exercice 35
from sklearn.base import BaseEstimator, TransformerMixin
import numpy as np

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) -> "FeatureEngineer":
        return self

    def transform(self, X: np.ndarray) -> np.ndarray:
        if self.add_polynomials and X.shape[1] >= 2:
            X = np.column_stack([X, X[:, 0] * X[:, 1]])
        return X

---

Exercices 36–38 : Architecture & CLI

Correction — Exercice 37 : CLI Typer

import typer
from rich.console import Console
from rich.table import Table
from pathlib import Path
from collections import Counter

app = typer.Typer()
console = Console()

@app.command()
def analyze(
    file: Path = typer.Argument(..., help="Path to log file"),
    level: str = typer.Option("ERROR", "--level", "-l"),
    top: int = typer.Option(10, "--top", "-t"),
) -> None:
    if not file.exists():
        console.print(f"[red]File {file} not found[/red]")
        raise typer.Exit(1)

    counter: Counter[str] = Counter()
    with open(file, encoding="utf-8") as f:
        for line in f:
            if level in line:
                module = line.split()[2] if len(line.split()) > 2 else "unknown"
                counter[module] += 1

    table = Table(title=f"Top {top} modules with {level} errors")
    table.add_column("Module", style="cyan")
    table.add_column("Count", style="magenta")

    for module, count in counter.most_common(top):
        table.add_row(module, str(count))

    console.print(table)

if __name__ == "__main__":
    app()

---

Exercices 39–40 : Docker

Correction — Exercice 39 : Dockerfile multi-stage

FROM python:3.13-slim AS builder
WORKDIR /build
COPY pyproject.toml poetry.lock ./
RUN pip install poetry && poetry export -f requirements.txt -o requirements.txt
RUN pip install --prefix=/install -r requirements.txt

FROM python:3.13-slim
COPY --from=builder /install /usr/local
COPY . /app
WORKDIR /app
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0"]

---

Corrections QCM

Toutes les réponses aux QCM du chapitre 18 sont listées ci-dessous :