Ce TP final du niveau Expert mobilise toutes les notions avancées : regex, générateurs, décorateurs, POO, exceptions, fichiers, type hints, tests unitaires. Vous allez construire un analyseur de logs digne d’un outil pro — utilisable en vrai.
🎯 Objectifs pédagogiques
Parser un format de log texte avec des regex.
Utiliser des générateurs pour traiter des fichiers potentiellement énormes (lazy).
Créer des classes typées avec @dataclass et type hints.
Écrire des décorateurs (chrono, logging).
Tester unitairement chaque composant.
Produire un rapport structuré en CLI.
Temps estimé : 3 h 30.
49.1 Contexte
Vous héritez d’un fichier access.log produit par un serveur web. Chaque ligne ressemble à :
2026-04-22 14:30:12 INFO GET /api/users 200 42ms
2026-04-22 14:30:15 WARN POST /login 401 15ms
2026-04-22 14:30:18 ERROR GET /api/orders 500 3200ms
Format : date heure NIVEAU METHODE URL CODE DUREEms.
Vous devez :
Lire ce log (potentiellement énorme : plusieurs Go).
Parser chaque ligne en objet structuré.
Filtrer, agréger, analyser les données.
Produire un rapport (top URLs, distribution des codes, erreurs les plus fréquentes).
Méthode
Un fichier par module : parseur.py, rapport.py, cli.py, tests/.
Testez chaque fonction avant de passer à la suite.
Typez tout le code public — vérifiable avec mypy --strict.
Utilisez les générateurs pour ne rien charger en mémoire inutilement.
Partie 1 — La classe LogEntry
Consignes
Créez une @dataclass typée représentant une ligne de log.
horodatage: datetime
niveau: Literal["DEBUG", "INFO", "WARN", "ERROR"]
methode: str
url: str
code: int
duree_ms: int
Ajoutez des properties pour :
est_erreur : True si code >= 400.
est_lent : True si duree_ms > 1000.
Exception personnalisée LigneInvalideError pour signaler un parsing qui échoue.
🔍 Solution Partie 1
from dataclasses import dataclassfrom datetime import datetimefrom typing import LiteralNiveauLog = Literal["DEBUG", "INFO", "WARN", "ERROR"]class LigneInvalideError(Exception):"""Levée quand une ligne de log ne peut pas être parsée."""pass@dataclassclass LogEntry: horodatage: datetime niveau: NiveauLog methode: str url: str code: int duree_ms: int@propertydef est_erreur(self) ->bool:returnself.code >=400@propertydef est_lent(self) ->bool:returnself.duree_ms >1000def__str__(self) ->str: emoji ="🔴"ifself.est_erreur else"🟡"ifself.est_lent else"🟢"return (f"{emoji}{self.horodatage:%Y-%m-%d %H:%M:%S} "f"{self.niveau:<5}{self.methode}{self.url} "f"[{self.code}] {self.duree_ms}ms")# Testentree = LogEntry( horodatage=datetime(2026, 4, 22, 14, 30, 12), niveau="INFO", methode="GET", url="/api/users", code=200, duree_ms=42,)print(entree)print(f"Est erreur ? {entree.est_erreur}")print(f"Est lent ? {entree.est_lent}")erreur = LogEntry( horodatage=datetime(2026, 4, 22, 14, 30, 18), niveau="ERROR", methode="GET", url="/api/orders", code=500, duree_ms=3200,)print(erreur)print(f"Est erreur ? {erreur.est_erreur}")print(f"Est lent ? {erreur.est_lent}")
🟢 2026-04-22 14:30:12 INFO GET /api/users [200] 42ms
Est erreur ? False
Est lent ? False
🔴 2026-04-22 14:30:18 ERROR GET /api/orders [500] 3200ms
Est erreur ? True
Est lent ? True
Points à retenir
Literal["DEBUG", ...] restreint les valeurs acceptées de niveau (mypy le vérifie).
@property pour des attributs calculés à la volée — sans stockage.
Exception personnalisée finissant par Error (convention PEP 8).
Partie 2 — Parser une ligne avec regex
Consignes
Écrivez parser_ligne(ligne: str) -> LogEntry qui extrait les champs d’une ligne.
Format à matcher :
YYYY-MM-DD HH:MM:SS NIVEAU METHODE URL CODE DUREEms
Levez LigneInvalideError si la ligne ne correspond pas au format.
Indice : utilisez re.compile pour performance, avec des groupes nommés(?P<nom>...).
🔍 Solution Partie 2
import refrom dataclasses import dataclassfrom datetime import datetimefrom typing import LiteralNiveauLog = Literal["DEBUG", "INFO", "WARN", "ERROR"]class LigneInvalideError(Exception):pass@dataclassclass LogEntry: horodatage: datetime niveau: NiveauLog methode: str url: str code: int duree_ms: int@propertydef est_erreur(self) ->bool:returnself.code >=400# Motif avec groupes nommésMOTIF_LOG = re.compile(r"(?P<date>\d{4}-\d{2}-\d{2})\s+"r"(?P<heure>\d{2}:\d{2}:\d{2})\s+"r"(?P<niveau>DEBUG|INFO|WARN|ERROR)\s+"r"(?P<methode>GET|POST|PUT|DELETE|PATCH)\s+"r"(?P<url>\S+)\s+"r"(?P<code>\d{3})\s+"r"(?P<duree>\d+)ms")def parser_ligne(ligne: str) -> LogEntry:"""Parse une ligne de log en LogEntry, ou lève LigneInvalideError.""" ligne = ligne.strip() match = MOTIF_LOG.match(ligne)if match isNone:raise LigneInvalideError(f"Ligne invalide : {ligne!r}")try: horodatage = datetime.strptime(f"{match['date']}{match['heure']}","%Y-%m-%d %H:%M:%S" )exceptValueErroras e:raise LigneInvalideError(f"Date invalide : {e}") from ereturn LogEntry( horodatage=horodatage, niveau=match["niveau"], # type: ignore[arg-type] methode=match["methode"], url=match["url"], code=int(match["code"]), duree_ms=int(match["duree"]), )# Testslignes_valides = ["2026-04-22 14:30:12 INFO GET /api/users 200 42ms","2026-04-22 14:30:15 WARN POST /login 401 15ms","2026-04-22 14:30:18 ERROR GET /api/orders 500 3200ms",]for l in lignes_valides: entry = parser_ligne(l)print(f" {entry.methode:<4}{entry.url:<15} [{entry.code}]")
GET /api/users [200]
POST /login [401]
GET /api/orders [500]
Test des cas invalides
lignes_invalides = ["pas du tout un log","2026-04-22 14:30:12 INFO", # incomplet"2026-99-99 14:30:12 INFO GET /x 200 10ms", # date invalide]for l in lignes_invalides:try: parser_ligne(l)except LigneInvalideError as e:print(f" ❌ {l[:30]}... → {e}")
❌ pas du tout un log... → Ligne invalide : 'pas du tout un log'
❌ 2026-04-22 14:30:12 INFO... → Ligne invalide : '2026-04-22 14:30:12 INFO'
❌ 2026-99-99 14:30:12 INFO GET /... → Date invalide : time data '2026-99-99 14:30:12' does not match format '%Y-%m-%d %H:%M:%S'
Points à retenir
Groupes nommés(?P<nom>...) : accès par nom plutôt qu’indice.
match[\"nom\"] : équivalent à match.group(\"nom\").
re.compile une seule fois : plus rapide si la fonction est appelée souvent.
raise ... from e : préserve la chaîne d’exceptions pour le debug.
Le type: ignore[arg-type] est un petit compromis : mypy ne peut pas prouver que la regex renvoie un des 4 niveaux valides — on le sait, lui pas.
Partie 3 — Lire un fichier en générateur
Consignes
Écrivez lire_logs(chemin: Path) -> Iterator[LogEntry] qui :
Ouvre le fichier proprement avec with.
Itère ligne par ligne (pas readlines !).
Ignore les lignes vides.
Saute les lignes invalides en affichant un avertissement sur stderr.
Renvoie les LogEntryun par un via yield.
Indice : utilisez from collections.abc import Iterator pour l’annotation.
🔍 Solution Partie 3
import reimport sysfrom collections.abc import Iteratorfrom dataclasses import dataclassfrom datetime import datetimefrom pathlib import Pathfrom typing import LiteralNiveauLog = Literal["DEBUG", "INFO", "WARN", "ERROR"]class LigneInvalideError(Exception): pass@dataclassclass LogEntry: horodatage: datetime niveau: NiveauLog methode: str url: str code: int duree_ms: int@propertydef est_erreur(self) ->bool:returnself.code >=400MOTIF_LOG = re.compile(r"(?P<date>\d{4}-\d{2}-\d{2})\s+"r"(?P<heure>\d{2}:\d{2}:\d{2})\s+"r"(?P<niveau>DEBUG|INFO|WARN|ERROR)\s+"r"(?P<methode>GET|POST|PUT|DELETE|PATCH)\s+"r"(?P<url>\S+)\s+"r"(?P<code>\d{3})\s+"r"(?P<duree>\d+)ms")def parser_ligne(ligne: str) -> LogEntry: ligne = ligne.strip() match = MOTIF_LOG.match(ligne)if match isNone:raise LigneInvalideError(f"Ligne invalide : {ligne!r}") horodatage = datetime.strptime(f"{match['date']}{match['heure']}", "%Y-%m-%d %H:%M:%S" )return LogEntry( horodatage=horodatage, niveau=match["niveau"], # type: ignore[arg-type] methode=match["methode"], url=match["url"], code=int(match["code"]), duree_ms=int(match["duree"]), )def lire_logs(chemin: Path, verbose: bool=False) -> Iterator[LogEntry]:"""Générateur d'entrées depuis un fichier log. Les lignes vides sont ignorées, les lignes invalides sont signalées sur stderr mais ne stoppent pas la lecture. """withopen(chemin, "r", encoding="utf-8") as f:for numero, ligne inenumerate(f, start=1):ifnot ligne.strip():continuetry:yield parser_ligne(ligne)except LigneInvalideError as e:if verbose:print(f"[Ligne {numero}] ignorée : {e}", file=sys.stderr)# Créer un fichier de testchemin_test = Path("access.log")contenu ="""2026-04-22 14:30:12 INFO GET /api/users 200 42ms2026-04-22 14:30:15 WARN POST /login 401 15ms2026-04-22 14:30:18 ERROR GET /api/orders 500 3200msligne_corrompue2026-04-22 14:30:22 INFO GET /api/products 200 78ms2026-04-22 14:30:25 ERROR POST /api/checkout 500 250ms"""chemin_test.write_text(contenu, encoding="utf-8")# Utilisation : lazy, rien n'est stocké en mémoirefor entry in lire_logs(chemin_test): statut ="🔴"if entry.est_erreur else"🟢"print(f" {statut}{entry.methode}{entry.url} [{entry.code}]")# Nettoyagechemin_test.unlink()
🟢 GET /api/users [200]
🔴 POST /login [401]
🔴 GET /api/orders [500]
🟢 GET /api/products [200]
🔴 POST /api/checkout [500]
Points à retenir
Iterator[LogEntry] (ou Generator[LogEntry, None, None]) pour typer un générateur.
with open(...) as f: garantit la fermeture.
for ligne in f: : lecture streamée, mémoire constante.
Résilience : on continue même si une ligne est invalide (avec log sur stderr).
enumerate(f, start=1) pour numéroter les lignes depuis 1.
file=sys.stderr pour les messages d’erreur (séparés de stdout).
Partie 4 — Décorateur de chronométrage
Consignes
Écrivez un décorateur @chrono qui :
Mesure le temps d’exécution de la fonction.
Affiche "{fonction} terminée en XX ms" sur stderr.
Préserve le nom et la docstring avec functools.wraps.
Est générique : fonctionne sur n’importe quelle fonction (*args, **kwargs).
Appliquez-le à une fonction d’analyse.
🔍 Solution Partie 4
import timeimport sysfrom functools import wrapsfrom typing import Callable, TypeVar, ParamSpecfrom collections.abc import Iteratorfrom dataclasses import dataclassfrom datetime import datetimefrom pathlib import Pathfrom typing import LiteralP = ParamSpec("P")R = TypeVar("R")def chrono(fonction: Callable[P, R]) -> Callable[P, R]:"""Chronomètre une fonction et affiche la durée."""@wraps(fonction)def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: debut = time.perf_counter() resultat = fonction(*args, **kwargs) duree = (time.perf_counter() - debut) *1000print(f"[chrono] {fonction.__name__} terminée en {duree:.1f} ms",file=sys.stderr)return resultatreturn wrapper# Préparer un fichier de test plus grosNiveauLog = Literal["DEBUG", "INFO", "WARN", "ERROR"]@dataclassclass LogEntry: horodatage: datetime niveau: NiveauLog methode: str url: str code: int duree_ms: int@propertydef est_erreur(self) ->bool:returnself.code >=400import reMOTIF_LOG = re.compile(r"(?P<date>\d{4}-\d{2}-\d{2})\s+"r"(?P<heure>\d{2}:\d{2}:\d{2})\s+"r"(?P<niveau>DEBUG|INFO|WARN|ERROR)\s+"r"(?P<methode>GET|POST|PUT|DELETE|PATCH)\s+"r"(?P<url>\S+)\s+"r"(?P<code>\d{3})\s+"r"(?P<duree>\d+)ms")class LigneInvalideError(Exception): passdef parser_ligne(ligne: str) -> LogEntry: ligne = ligne.strip() match = MOTIF_LOG.match(ligne)if match isNone:raise LigneInvalideError(f"Ligne invalide") horodatage = datetime.strptime(f"{match['date']}{match['heure']}", "%Y-%m-%d %H:%M:%S" )return LogEntry( horodatage=horodatage, niveau=match["niveau"], # type: ignore[arg-type] methode=match["methode"], url=match["url"], code=int(match["code"]), duree_ms=int(match["duree"]), )def lire_logs(chemin: Path) -> Iterator[LogEntry]:withopen(chemin, "r", encoding="utf-8") as f:for ligne in f:ifnot ligne.strip():continuetry:yield parser_ligne(ligne)except LigneInvalideError:continue# Créer un fichier de 5000 lignes pour voir le chrono travaillerimport randomrandom.seed(42)chemin_test = Path("gros.log")niveaux = ["DEBUG", "INFO", "WARN", "ERROR"]methodes = ["GET", "POST", "PUT", "DELETE"]urls = ["/api/users", "/api/orders", "/login", "/api/products", "/api/checkout"]withopen(chemin_test, "w", encoding="utf-8") as f:for i inrange(5000): h =10+ i //600 m = (i //10) %60 s = i %60 niveau = random.choice(niveaux) methode = random.choice(methodes) url = random.choice(urls) code = random.choice([200, 200, 200, 401, 404, 500]) duree = random.randint(10, 5000) f.write(f"2026-04-22 {h:02d}:{m:02d}:{s:02d}{niveau:<5} "f"{methode}{url}{code}{duree}ms\n")@chronodef compter_toutes_entrees(chemin: Path) ->int:"""Compte le nombre d'entrées valides dans un log."""returnsum(1for _ in lire_logs(chemin))@chronodef compter_erreurs(chemin: Path) ->int:"""Compte les erreurs (code >= 400)."""returnsum(1for e in lire_logs(chemin) if e.est_erreur)# Testsn_total = compter_toutes_entrees(chemin_test)n_err = compter_erreurs(chemin_test)print(f"\nTotal : {n_total}")print(f"Erreurs : {n_err}")print(f"Taux d'erreur : {n_err / n_total *100:.1f} %")# Nettoyagechemin_test.unlink()
Messages d’erreur sur stderr, rapport sur stdout — permet de rediriger séparément.
if __name__ == "__main__": pour que le script soit importable sans se lancer.
🏆 Bonus — Pour aller plus loin
Export CSV : exporter_csv(analyseur, chemin) avec le module csv.
Filtre temporel : méthode filtrer_periode(debut, fin) qui renvoie un nouveau AnalyseurLogs.
Détection d’anomalies : méthode qui renvoie les URLs avec un taux d’erreur supérieur à la moyenne (sur 3 écarts-types).
Mode streaming : ne pas charger toutes les entrées en mémoire — utiliser des générateurs pour les stats. Défi : comment faire top_urls sans tout stocker ? (Indice : un seul passage + Counter).
Tests pytest avec paramétrisation et fixtures : réécrire les tests en pytest.
Rapport HTML : générer un rapport au format HTML stylisé avec CSS.
Graphiques : utiliser matplotlib pour visualiser l’évolution des erreurs dans le temps.
🔍 Idée pour le bonus 3 — Détection d’anomalies
import statisticsdef urls_anormales(analyseur, seuil_sigma: float=3.0) ->list[str]:"""Renvoie les URLs dont le taux d'erreur dépasse moyenne + sigma × écart-type."""from collections import defaultdict stats: dict[str, list[bool]] = defaultdict(list)for e in analyseur.entrees: stats[e.url].append(e.est_erreur) taux = {url: sum(v) /len(v) for url, v in stats.items()}iflen(taux) <2:return [] moyenne = statistics.mean(taux.values()) ecart = statistics.stdev(taux.values()) seuil = moyenne + seuil_sigma * ecartreturn [url for url, t in taux.items() if t > seuil]
Pattern réel : alerte opérationnelle sur les endpoints qui dégradent.
✅ Validation du niveau Expert
Bilan du TP
Si vous avez résolu les 7 parties (même partiellement), vous maîtrisez les compétences du niveau Expert TOSA :
✅ Regex avec groupes nommés et re.compile
✅ Générateurs pour traiter des flux (pas de chargement mémoire)
✅ Décorateurs typés et génériques
✅ Closures (dans les décorateurs)
✅ POO avancée avec @dataclass, properties, classmethods
✅ Exceptions personnalisées avec chaînage (raise ... from)
✅ pathlib pour les chemins
✅ Type hints complets + Literal, Iterator, Iterable, forward references