--- jupytext: text_representation: extension: .md format_name: myst format_version: 0.13 jupytext_version: 1.16.0 kernelspec: name: python3 display_name: Python 3 --- # Gestionnaires de contexte ```{code-cell} python :tags: [hide-input] import matplotlib.pyplot as plt import matplotlib.patches as patches import numpy as np import seaborn as sns sns.set_theme(style="whitegrid", palette="muted", font_scale=1.1) ``` ## Le problème de la gestion des ressources Toute application sérieuse manipule des **ressources** : fichiers ouverts sur le disque, connexions à une base de données, verrous de synchronisation entre fils d'exécution, sockets réseau, handles vers des périphériques matériels. Ces ressources ont un point commun fondamental : elles doivent être **libérées** lorsqu'on n'en a plus besoin. Un fichier non fermé peut corrompre des données ou épuiser le quota de descripteurs de fichiers du système d'exploitation. Une connexion à une base de données non libérée peut bloquer d'autres clients. Un verrou non relâché peut provoquer un interblocage (*deadlock*) fatal pour l'application. Le code naïf gère ces ressources de façon linéaire : on ouvre le fichier, on l'utilise, on le ferme. Mais cette approche ignore la réalité des programmes : les **exceptions**. Si une erreur survient après l'ouverture du fichier mais avant sa fermeture, le code de nettoyage n'est jamais exécuté. ```python # Code fragile : si une exception se produit lors de la lecture, # fichier.close() n'est jamais appelé. fichier = open("donnees.txt", "r") contenu = fichier.read() # Peut lever UnicodeDecodeError, IOError, etc. fichier.close() # Cette ligne peut ne jamais s'exécuter ! ``` La solution classique avant Python 2.5 consistait à utiliser `try / finally` pour garantir le nettoyage : ```python fichier = open("donnees.txt", "r") try: contenu = fichier.read() finally: fichier.close() # Exécuté même si une exception est levée ``` Ce code est correct, mais verbeux. Il est aussi source d'erreurs si l'on oublie le bloc `finally`. Multiplié sur des dizaines de ressources dans une application réelle, ce patron devient rapidement illisible. C'est précisément ce problème que l'instruction `with` résout de façon élégante. ## L'instruction `with` L'instruction `with` a été introduite dans Python 2.5 (PEP 343) pour encapsuler le patron `try / finally` dans une syntaxe propre et expressive. Elle s'appuie sur le protocole des **gestionnaires de contexte** (*context managers*). ```python with open("donnees.txt", "r") as fichier: contenu = fichier.read() # fichier.close() est appelé automatiquement ici, # que le bloc se soit terminé normalement ou avec une exception. ``` La syntaxe générale est la suivante : ```python with expression as variable: # Corps du bloc ... ``` `expression` doit retourner un objet qui implémente le protocole des gestionnaires de contexte. `variable` (la clause `as` est optionnelle) reçoit la valeur retournée par `__enter__`. À la sortie du bloc, qu'elle soit normale ou provoquée par une exception, la méthode `__exit__` est appelée systématiquement. On peut également gérer **plusieurs ressources** dans un seul `with`, ce qui était auparavant impossible sans imbriquer les blocs : ```python # Depuis Python 3.10, on peut utiliser des parenthèses pour la lisibilité with ( open("source.txt", "r") as source, open("destination.txt", "w") as dest, ): dest.write(source.read()) ``` Le flux d'exécution exact est illustré dans la visualisation ci-dessous : ```{code-cell} python :tags: [hide-input] fig, ax = plt.subplots(figsize=(12, 9)) ax.set_xlim(0, 12) ax.set_ylim(0, 10) ax.axis('off') ax.set_title("Flux d'exécution d'un bloc with", fontsize=15, fontweight='bold', pad=15) c_blue = '#3498db' c_green = '#27ae60' c_red = '#e74c3c' c_orange = '#e67e22' c_gray = '#7f8c8d' c_dark = '#2c3e50' def box(ax, x, y, w, h, color, text, fontsize=10, text_color='white', alpha=0.92): rect = patches.FancyBboxPatch( (x - w/2, y - h/2), w, h, boxstyle="round,pad=0.12", linewidth=2, edgecolor=color, facecolor=color, alpha=alpha ) ax.add_patch(rect) ax.text(x, y, text, ha='center', va='center', fontsize=fontsize, fontweight='bold', color=text_color, wrap=True) def arrow(ax, x1, y1, x2, y2, color=c_dark, label='', label_side='right'): ax.annotate('', xy=(x2, y2), xytext=(x1, y1), arrowprops=dict(arrowstyle='->', color=color, lw=2.0)) if label: mx, my = (x1 + x2) / 2, (y1 + y2) / 2 offset = 0.35 if label_side == 'right' else -0.35 ax.text(mx + offset, my, label, ha='center', va='center', fontsize=8, color=color, style='italic') # Noeuds du diagramme de flux box(ax, 6, 9.2, 4.5, 0.8, c_blue, "with expression as var :") box(ax, 6, 7.8, 4.5, 0.8, c_green, "Appel de __enter__()\n→ valeur liée à var", fontsize=9) box(ax, 6, 6.3, 4.5, 0.9, c_dark, "Exécution du corps du bloc", fontsize=10) # Losange décision diamond_x, diamond_y = 6, 4.8 diamond = patches.FancyBboxPatch( (diamond_x - 2.0, diamond_y - 0.55), 4.0, 1.1, boxstyle="round,pad=0.1", linewidth=2, edgecolor=c_orange, facecolor=c_orange, alpha=0.85 ) ax.add_patch(diamond) ax.text(diamond_x, diamond_y, "Exception levée ?", ha='center', va='center', fontsize=10, fontweight='bold', color='white') # Chemin Normal (gauche) box(ax, 2.5, 3.2, 3.2, 0.8, c_green, "__exit__(None, None, None)", fontsize=8.5) box(ax, 2.5, 2.0, 3.2, 0.8, c_green, "Sortie normale ✓", fontsize=9) # Chemin Exception (droite) box(ax, 9.5, 3.2, 3.2, 0.8, c_red, "__exit__(type, val, tb)", fontsize=8.5) # Sous-décision : exception supprimée ? sub_diamond = patches.FancyBboxPatch( (9.5 - 2.0, 2.0 - 0.45), 4.0, 0.9, boxstyle="round,pad=0.08", linewidth=1.5, edgecolor=c_orange, facecolor=c_orange, alpha=0.7 ) ax.add_patch(sub_diamond) ax.text(9.5, 2.0, "__exit__ retourne True ?", ha='center', va='center', fontsize=8, fontweight='bold', color='white') box(ax, 7.5, 0.8, 2.6, 0.7, c_green, "Exception supprimée ✓", fontsize=8) box(ax, 11.0, 0.8, 2.2, 0.7, c_red, "Propagation ✗", fontsize=8) # Flèches arrow(ax, 6, 8.8, 6, 8.2) arrow(ax, 6, 7.4, 6, 6.75) arrow(ax, 6, 5.85, 6, 5.35) # Non -> gauche ax.annotate('', xy=(2.5, 3.6), xytext=(4.0, 4.8), arrowprops=dict(arrowstyle='->', color=c_green, lw=2.0)) ax.text(2.8, 4.4, 'Non', fontsize=9, color=c_green, fontweight='bold') # Oui -> droite ax.annotate('', xy=(9.5, 3.6), xytext=(8.0, 4.8), arrowprops=dict(arrowstyle='->', color=c_red, lw=2.0)) ax.text(9.2, 4.4, 'Oui', fontsize=9, color=c_red, fontweight='bold') arrow(ax, 2.5, 2.8, 2.5, 2.4) arrow(ax, 9.5, 2.8, 9.5, 2.45) ax.annotate('', xy=(7.5, 1.15), xytext=(8.5, 2.0), arrowprops=dict(arrowstyle='->', color=c_green, lw=1.8)) ax.text(7.6, 1.65, 'Oui', fontsize=8, color=c_green, fontweight='bold') ax.annotate('', xy=(11.0, 1.15), xytext=(10.5, 2.0), arrowprops=dict(arrowstyle='->', color=c_red, lw=1.8)) ax.text(11.3, 1.65, 'Non', fontsize=8, color=c_red, fontweight='bold') plt.tight_layout() plt.show() ``` ## Implémenter `__enter__` et `__exit__` N'importe quel objet peut devenir un gestionnaire de contexte en implémentant deux méthodes spéciales : `__enter__` et `__exit__`. Ce protocole est défini dans la PEP 343. **`__enter__(self)`** est appelée au moment de l'entrée dans le bloc `with`. Sa valeur de retour est liée à la variable de la clause `as`. Elle peut retourner `self`, un objet différent, ou `None`. **`__exit__(self, exc_type, exc_val, exc_tb)`** est appelée à la sortie du bloc, avec trois arguments : - `exc_type` : la classe de l'exception levée, ou `None` si aucune exception. - `exc_val` : l'instance de l'exception, ou `None`. - `exc_tb` : l'objet traceback, ou `None`. Si `__exit__` retourne une valeur vraie (*truthy*), l'exception est **supprimée** et l'exécution continue après le bloc `with`. Si elle retourne `False` ou `None`, l'exception est propagée normalement. ```{code-cell} python class Chronometre: """Gestionnaire de contexte pour mesurer le temps d'exécution.""" import time def __enter__(self): import time self._debut = time.perf_counter() print("Chronomètre démarré.") return self # On retourne self pour permettre cm.duree def __exit__(self, exc_type, exc_val, exc_tb): import time self.duree = time.perf_counter() - self._debut if exc_type is None: print(f"Temps écoulé : {self.duree:.4f} s") else: print(f"Exception {exc_type.__name__} après {self.duree:.4f} s") return False # On ne supprime pas l'exception with Chronometre() as cm: total = sum(range(1_000_000)) print(f"Somme : {total}, durée : {cm.duree:.4f} s") ``` ```{code-cell} python class SupprimerErreur: """Supprime une exception spécifique dans le bloc.""" def __init__(self, *types): self._types = types def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is not None and issubclass(exc_type, self._types): print(f"Exception {exc_type.__name__} supprimée : {exc_val}") return True # Supprime l'exception return False with SupprimerErreur(ZeroDivisionError): resultat = 1 / 0 # Serait normalement fatale print("Cette ligne n'est pas atteinte") print("Exécution continue après la suppression de l'exception.") ``` ```{prf:definition} Protocole des gestionnaires de contexte :label: definition-12-01 Un **gestionnaire de contexte** est tout objet qui implémente les méthodes `__enter__` et `__exit__`. L'instruction `with` appelle `__enter__` à l'entrée du bloc et `__exit__` à la sortie, qu'elle soit normale ou provoquée par une exception. Si `__exit__` retourne une valeur vraie, l'exception éventuelle est supprimée. ``` ```{prf:remark} :label: remark-12-01 La signature complète de `__exit__` — `(self, exc_type, exc_val, exc_tb)` — permet d'inspecter précisément l'exception. On peut ainsi décider de ne supprimer que certaines exceptions (par exemple, uniquement `FileNotFoundError`), de journaliser le traceback, ou de transformer une exception en une autre en la levant explicitement dans `__exit__`. ``` ## `contextlib` Le module `contextlib` de la bibliothèque standard offre plusieurs utilitaires pour créer et utiliser des gestionnaires de contexte sans avoir à écrire une classe complète. ### `@contextmanager` Le décorateur `@contextmanager` transforme une fonction génératrice en gestionnaire de contexte. La partie avant le `yield` correspond à `__enter__`, la valeur du `yield` est liée à la clause `as`, et la partie après le `yield` correspond à `__exit__`. ```{code-cell} python from contextlib import contextmanager @contextmanager def gestion_transaction(connexion_simulee: str): """Simule une transaction de base de données.""" print(f"BEGIN TRANSACTION sur {connexion_simulee}") try: yield connexion_simulee # La valeur liée à `as` print("COMMIT") except Exception as e: print(f"ROLLBACK à cause de : {e}") raise # On re-lève l'exception après le rollback with gestion_transaction("db_principale") as conn: print(f"Exécution de requêtes sur {conn}") # Simulons une opération réussie ``` ```{code-cell} python # Exemple avec une exception try: with gestion_transaction("db_secondaire") as conn: print(f"Tentative d'écriture sur {conn}") raise ValueError("Contrainte de clé étrangère violée") except ValueError as e: print(f"Erreur capturée en dehors : {e}") ``` ### `contextlib.suppress` `contextlib.suppress` est l'équivalent standard de notre `SupprimerErreur` ci-dessus. Il supprime proprement une ou plusieurs classes d'exceptions : ```{code-cell} python import contextlib import os # Supprimer FileNotFoundError si le fichier n'existe pas with contextlib.suppress(FileNotFoundError): os.remove("/tmp/fichier_inexistant_12345.txt") print("Pas d'erreur levée, même si le fichier n'existait pas.") ``` ### `contextlib.redirect_stdout` `redirect_stdout` redirige temporairement la sortie standard vers un objet de type fichier, ce qui est très utile pour capturer la sortie de fonctions tierces : ```{code-cell} python import io import contextlib sortie = io.StringIO() with contextlib.redirect_stdout(sortie): print("Ce message va dans le buffer, pas dans la console.") print("Pareil pour celui-ci.") contenu_capture = sortie.getvalue() print(f"Contenu capturé ({len(contenu_capture)} caractères) :") print(repr(contenu_capture)) ``` ### `contextlib.ExitStack` `ExitStack` est l'outil le plus puissant de `contextlib`. Il permet de gérer un **nombre dynamique** de gestionnaires de contexte, déterminé à l'exécution plutôt qu'à l'écriture du code : ```{code-cell} python import contextlib fichiers_a_ouvrir = ["fichier_a.txt", "fichier_b.txt", "fichier_c.txt"] # Créer les fichiers de test for nom in fichiers_a_ouvrir: with open(f"/tmp/{nom}", "w") as f: f.write(f"Contenu de {nom}\n") with contextlib.ExitStack() as stack: fichiers = [ stack.enter_context(open(f"/tmp/{nom}", "r")) for nom in fichiers_a_ouvrir ] for f in fichiers: print(f.readline().strip()) # Tous les fichiers sont fermés ici, même si une exception survient. ``` ```{prf:example} Quand utiliser ExitStack ? :label: example-12-01 `ExitStack` est indispensable quand le nombre de ressources à gérer n'est connu qu'à l'exécution : lecture d'une liste de fichiers dont le nombre varie, ouverture conditionnelle de ressources selon des paramètres de configuration, composition de plugins qui exposent chacun un gestionnaire de contexte. Sans `ExitStack`, il faudrait imbriquer manuellement des blocs `with`, ce qui est impossible quand le nombre est variable. ``` ## Gestionnaires de contexte asynchrones Avec l'essor de la programmation asynchrone en Python (`asyncio`), le protocole des gestionnaires de contexte a été étendu aux contextes asynchrones. L'instruction `async with` repose sur deux méthodes coroutines : `__aenter__` et `__aexit__`. ```python import asyncio class ConnexionAsync: async def __aenter__(self): print("Ouverture de la connexion asynchrone...") await asyncio.sleep(0.01) # Simule une opération I/O return self async def __aexit__(self, exc_type, exc_val, exc_tb): print("Fermeture de la connexion asynchrone...") await asyncio.sleep(0.01) return False async def main(): async with ConnexionAsync() as conn: print(f"Utilisation de la connexion : {conn}") asyncio.run(main()) ``` Le module `contextlib` propose également `@asynccontextmanager` pour créer des gestionnaires de contexte asynchrones à partir d'une fonction génératrice asynchrone (`async def` avec `yield`) : ```python from contextlib import asynccontextmanager @asynccontextmanager async def ressource_async(nom: str): print(f"Acquisition de {nom}...") await asyncio.sleep(0.01) try: yield nom finally: print(f"Libération de {nom}...") await asyncio.sleep(0.01) ``` ```{prf:remark} :label: remark-12-02 Les gestionnaires de contexte asynchrones ne peuvent être utilisés qu'à l'intérieur de coroutines (`async def`). Ils sont essentiels pour les bibliothèques qui gèrent des ressources I/O asynchrones : connexions HTTP (`aiohttp`), connexions aux bases de données asynchrones (`asyncpg`, `databases`), ou verrous asynchrones (`asyncio.Lock`). ``` ## Cas d'usage courants Les gestionnaires de contexte s'appliquent à une variété impressionnante de situations concrètes. En voici les plus représentatives. **Fichiers et flux.** C'est l'usage le plus connu. `open()` retourne un gestionnaire de contexte qui ferme le fichier à la sortie du bloc. Les bibliothèques `zipfile`, `tarfile`, `sqlite3` et `csv` exposent des interfaces similaires. **Verrous de synchronisation.** `threading.Lock`, `threading.RLock`, `asyncio.Lock` et leurs dérivés implémentent tous le protocole. L'instruction `with verrou:` garantit que le verrou est toujours relâché, évitant les interblocages : ```{code-cell} python import threading compteur = 0 verrou = threading.Lock() def incrementer(n: int) -> None: global compteur for _ in range(n): with verrou: # Acquisition et libération garanties compteur += 1 threads = [threading.Thread(target=incrementer, args=(10_000,)) for _ in range(5)] for t in threads: t.start() for t in threads: t.join() print(f"Compteur final : {compteur} (attendu : 50 000)") ``` **Transactions de base de données.** Le gestionnaire de contexte de `sqlite3` confirme (*commit*) ou annule (*rollback*) automatiquement la transaction. **Chronomètres de performance.** Le module `contextlib` combiné avec `time.perf_counter` permet de mesurer précisément la durée d'un bloc de code. **Tests.** `pytest` et `unittest` utilisent des gestionnaires de contexte pour vérifier qu'une exception est bien levée (`pytest.raises`, `unittest.assertRaises`), pour patcher temporairement des objets (`unittest.mock.patch`), ou pour capturer des avertissements. ```{code-cell} python import contextlib import io # Exemple : capturer les avertissements dans un contexte de test import warnings with warnings.catch_warnings(record=True) as w: warnings.simplefilter("always") warnings.warn("Ceci est un avertissement de démonstration.", DeprecationWarning) assert len(w) == 1 assert issubclass(w[0].category, DeprecationWarning) print(f"Avertissement capturé : {w[0].message}") ``` ## Résumé Dans ce chapitre, nous avons exploré les **gestionnaires de contexte**, un mécanisme fondamental de Python pour la gestion propre des ressources. - Le problème central est la **libération garantie des ressources** même en cas d'exception. Le patron `try / finally` résout le problème mais est verbeux. - L'instruction **`with`** encapsule ce patron dans une syntaxe claire. Elle appelle `__enter__` à l'entrée et `__exit__` à la sortie du bloc, sans exception. - Le **protocole** repose sur `__enter__` (qui peut retourner une valeur) et `__exit__(exc_type, exc_val, exc_tb)` (qui peut supprimer une exception en retournant `True`). - Le module **`contextlib`** simplifie la création de gestionnaires de contexte : `@contextmanager` pour les fonctions génératrices, `suppress` pour ignorer des exceptions, `redirect_stdout` pour capturer la sortie, et `ExitStack` pour gérer un nombre dynamique de ressources. - Les **gestionnaires de contexte asynchrones** (`async with`, `__aenter__`, `__aexit__`) étendent le protocole à la programmation asynchrone. - Les cas d'usage couvrent les fichiers, les verrous, les transactions, les chronomètres et les tests. Dans le chapitre suivant, nous approfondirons la **gestion des erreurs et des exceptions** : la hiérarchie complète des exceptions Python, la syntaxe `try / except / else / finally`, et les bonnes pratiques pour écrire du code robuste.