Context manager et l'instruction with en Python

1. Aperçu

Ouvrir un fichier, se connecter à une base de données ou acquérir un verrou nécessitent tous une étape de nettoyage à la fin. Cette étape doit s’exécuter même si le code au milieu lève une exception. Python résout cela avec les context managers et l’instruction with.

Cet article explique comment with fonctionne en dessous et comment écrire votre propre context manager avec une classe ou avec contextlib. Vous apprendrez également comment gérer les exceptions dans __exit__, comment tester les context managers et les erreurs courantes lors de leur utilisation.

2. Le problème : nettoyer les ressources avec try/finally

Voyons comment lire un fichier sans utiliser with :

f = open("orders.csv", encoding="utf-8")
try:
    header = f.readline()
    print(header)
finally:
    f.close()

Cette approche est correcte mais verbeux. C’est aussi facile à oublier. Si vous mettez open() à l’intérieur du try, et que open() échoue, alors f n’a pas été assigné et le bloc finally lèvera une NameError. L’instruction with encapsule l’intégralité du modèle ci-dessus :

with open("orders.csv", encoding="utf-8") as f:
    header = f.readline()
    print(header)

print(f.closed)

order_id,customer,amount

True

Quand vous quittez le bloc with, le fichier a été fermé. Cela est vrai que vous quittiez normalement ou avec une exception.

3. Le protocole du context manager

Tout objet ayant les deux méthodes __enter__ et __exit__ peut être utilisé avec with. Conceptuellement, l’instruction :

with EXPR as target:
    BODY

est équivalente au code suivant :

manager = EXPR
target = manager.__enter__()
try:
    BODY
except BaseException as e:
    if not manager.__exit__(type(e), e, e.__traceback__):
        raise
else:
    manager.__exit__(None, None, None)

Il y a trois points à retenir :

• La valeur assignée à target est le résultat de __enter__(), et non l’EXPR elle-même.
• __exit__ est toujours appelé. Lorsqu’une exception est levée, elle reçoit (exc_type, exc_value, traceback). Quand il n’y a pas d’erreur, les trois paramètres sont None.
• Si __exit__ retourne une valeur truthy, l’exception est supprimée. Sinon, l’exception est levée à nouveau.

3.1. Écrire un context manager avec une classe

Le premier exemple est un Timer qui mesure le temps d’exécution d’un bloc de code :

import time


class Timer:
    def __init__(self, label: str):
        self.label = label
        self.elapsed: float | None = None

    def __enter__(self):
        self._start = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc, tb):
        self.elapsed = time.perf_counter() - self._start
        print(f"[{self.label}] {self.elapsed:.3f}s")
        return False


with Timer("somme_commandes") as t:
    total = sum(range(5_000_000))

print(t.elapsed > 0)

[somme_commandes] 0.081s
True

Explication détaillée :

• __enter__ enregistre l’heure de début et retourne self. Ainsi t est exactement l’objet Timer.
• __exit__ calcule le temps écoulé, l’affiche à l’écran, puis retourne False. Les exceptions éventuelles ne seront pas supprimées.
• La variable t reste accessible après le bloc with. C’est parce que with ne crée pas une nouvelle portée.

3.2. Gérer les exceptions dans __exit__

Les context managers sont les plus utiles quand vous avez besoin de réagir différemment selon que le bloc de code réussit ou échoue. L’exemple ci-dessous est une transaction simple sur un registre (ledger) :

class Transaction:
    def __init__(self, ledger: list):
        self.ledger = ledger
        self.pending: list = []

    def __enter__(self):
        return self.pending

    def __exit__(self, exc_type, exc, tb):
        if exc_type is None:
            self.ledger.extend(self.pending)
            print(f"commit {len(self.pending)} enregistrements")
        else:
            print(f"rollback en raison de {exc_type.__name__}: {exc}")
        return False


ledger = []

with Transaction(ledger) as tx:
    tx.append({"order_id": 1, "amount": 250_000})
    tx.append({"order_id": 2, "amount": 120_000})

try:
    with Transaction(ledger) as tx:
        tx.append({"order_id": 3, "amount": 90_000})
        raise ValueError("solde insuffisant")
except ValueError:
    pass

print(len(ledger))

commit 2 enregistrements
rollback en raison de ValueError: solde insuffisant
2

Dans le deuxième bloc with, la commande n° 3 n’a pas été enregistrée dans le ledger. Puisque __exit__ retourne False, la ValueError est toujours levée et l’appelant décide comment la gérer.

Remarque : Ne retournez True depuis __exit__ que si vous intentionnellement voulez supprimer l’exception. Une méthode __exit__ qui retourne accidentellement self ou un autre objet truthy fera disparaître l’erreur sans aucune trace.

4. Plus concis avec @contextmanager

Écrire une classe entière juste pour une étape de configuration et de nettoyage est assez verbeux. Le module contextlib fournit le décorateur @contextmanager pour transformer un générateur en context manager :

from contextlib import contextmanager


class PaymentService:
    def connect(self):
        print("ouvrir la connexion à la passerelle de paiement")

    def charge(self, customer: str, amount: int):
        print(f"débiter {amount} du compte de {customer}")

    def disconnect(self):
        print("fermer la connexion")


@contextmanager
def payment_session(service: PaymentService):
    service.connect()
    try:
        yield service
    finally:
        service.disconnect()


with payment_session(PaymentService()) as payment:
    payment.charge("Lan", 300_000)

ouvrir la connexion à la passerelle de paiement
débiter 300000 du compte de Lan
fermer la connexion

Voici comment le décorateur mappe le générateur vers le protocole du context manager :

• Le code avant yield joue le rôle de __enter__.
• La valeur yielded est la valeur assignée à la variable après as.
• Le code après yield joue le rôle de __exit__. Si le bloc with lève une exception, cette exception est levée à nouveau au niveau de yield.

Puisque l’exception apparaît directement à yield, le bloc try/finally autour de yield est obligatoire. Sans lui, disconnect() ne s’exécuterait pas en cas d’erreur.

Pour gérer chaque type d’erreur séparément, vous pouvez ajouter except :

@contextmanager
def safe_charge(order_id: int):
    try:
        yield
    except ConnectionError as e:
        print(f"commande {order_id}: erreur réseau, nouvelle tentative ultérieurement ({e})")


with safe_charge(42):
    raise ConnectionError("timeout")

print("continuer le traitement")

commande 42: erreur réseau, nouvelle tentative ultérieurement (timeout)
continuer le traitement

Le générateur capture l’erreur mais ne la relève pas, donc l’exception est supprimée. Ce résultat équivaut à ce que __exit__ retourne True.

5. Les utilitaires intégrés dans contextlib

5.1. suppress : ignorer une exception spécifique

import os
from contextlib import suppress

with suppress(FileNotFoundError):
    os.remove("cache_orders.tmp")

print("terminé")

terminé

Le code ci-dessus est plus concis et plus explicite qu’un try: ... except FileNotFoundError: pass.

5.2. ExitStack : nombre de ressources inconnu à l’avance

Supposons que vous ayez besoin de fusionner plusieurs fichiers de factures, mais que le nombre de fichiers ne soit connu qu’à l’exécution :

from contextlib import ExitStack
from pathlib import Path


def merge_invoices(paths: list[Path], output: Path) -> None:
    with ExitStack() as stack:
        files = [stack.enter_context(open(p, encoding="utf-8")) for p in paths]
        with open(output, "w", encoding="utf-8") as out:
            for f in files:
                out.write(f.read())


for i in range(3):
    Path(f"invoice_{i}.txt").write_text(f"facture {i}\n", encoding="utf-8")

merge_invoices([Path(f"invoice_{i}.txt") for i in range(3)], Path("all.txt"))
print(Path("all.txt").read_text(encoding="utf-8"))

facture 0
facture 1
facture 2

enter_context appelle __enter__ et mémorise le __exit__ correspondant. Quand ExitStack se termine, les fichiers sont fermés dans l’ordre inverse d’ouverture. Si le troisième fichier échoue à s’ouvrir, les deux premiers fichiers sont toujours fermés correctement.

5.3. closing et nullcontext

• closing(obj) transforme un objet ayant seulement une méthode close() en context manager.
• nullcontext(value) est un context manager qui ne fait rien. C’est utile quand un paramètre peut ou ne peut pas être un context manager :

from contextlib import nullcontext
import threading

def update_stock(product: dict, qty: int, lock: threading.Lock | None = None):
    with lock if lock is not None else nullcontext():
        product["stock"] -= qty

product = {"sku": "AO-01", "stock": 10}
update_stock(product, 3)
update_stock(product, 2, threading.Lock())
print(product["stock"])

5

6. Plusieurs context managers dans une seule instruction with

Vous pouvez déclarer plusieurs context managers dans une seule instruction with. À partir de Python 3.10, la syntaxe utilisant des parenthèses permet une meilleure mise en forme :

with (
    open("orders.csv", encoding="utf-8") as src,
    open("orders_backup.csv", "w", encoding="utf-8") as dst,
):
    dst.write(src.read())

Cette syntaxe est équivalente à deux blocs with imbriqués. src est ouvert en premier et fermé en dernier.

7. Async context manager

Avec le code asyncio, le protocole correspondant utilise __aenter__ et __aexit__, utilisés avec async with. La version générateur est @asynccontextmanager :

import asyncio
from contextlib import asynccontextmanager


@asynccontextmanager
async def order_queue(name: str):
    print(f"connexion à la file {name}")
    await asyncio.sleep(0.01)
    try:
        yield [f"order-{i}" for i in range(3)]
    finally:
        await asyncio.sleep(0.01)
        print(f"fermeture file {name}")


async def main():
    async with order_queue("payments") as messages:
        for m in messages:
            print("traiter", m)

asyncio.run(main())

connexion à la file payments
traiter order-0
traiter order-1
traiter order-2
fermeture file payments

Remarque : Utiliser with régulier avec un async context manager lèvera une TypeError (ou AttributeError sur les versions plus anciennes de Python). L’inverse est aussi vrai. Ces deux protocoles sont distincts.

8. Tester les context managers

Les context managers doivent être testés à la fois pour le chemin réussi et le chemin d’erreur. Installez d’abord pytest :

pip install pytest

Mettez la classe Transaction de la section 3.2 dans un fichier transaction.py, puis écrivez les tests :

# test_transaction.py
import pytest
from transaction import Transaction


def test_commit_when_no_error():
    ledger = []
    with Transaction(ledger) as tx:
        tx.append({"order_id": 1, "amount": 100})

    assert ledger == [{"order_id": 1, "amount": 100}]


def test_rollback_and_propagate_error():
    ledger = []
    with pytest.raises(ValueError, match="solde"):
        with Transaction(ledger) as tx:
            tx.append({"order_id": 2, "amount": 999})
            raise ValueError("solde insuffisant")

    assert ledger == []

pytest -q

..                                                       [100%]
2 passed in 0.01s

Le deuxième test vérifie deux choses à la fois : les données sont annulées, et l’exception n’est pas supprimée. pytest.raises lui-même est aussi un context manager, un exemple familier du sujet de cet article.

9. Erreurs courantes et quand ne pas les utiliser

Oublier try/finally autour de yield. C’est l’erreur la plus courante avec @contextmanager :

@contextmanager
def broken_session(service):
    service.connect()
    yield service
    service.disconnect()  # ne s'exécute pas si le bloc with lève une exception

Utiliser une ressource après la fin du bloc with. La variable existe toujours, mais la ressource a été libérée :

with open("orders.csv", encoding="utf-8") as f:
    pass
f.read()

ValueError: I/O operation on closed file.

Faire yield plusieurs fois dans un générateur. Le générateur utilisé avec @contextmanager doit yield exactement une fois. Si vous faites yield plus d’une fois, Python lève RuntimeError: generator didn't stop.

Réutiliser un context manager créé avec @contextmanager. Cet objet ne peut être utilisé qu’une seule fois. Appeler with une deuxième fois sur le même objet lèvera une erreur. Chaque fois que vous l’utilisez, appelez à nouveau la fonction pour créer un nouvel objet.

Abus d’utilisation. Vous ne devriez pas utiliser de context managers dans les cas suivants :

• La ressource doit persister plus longtemps qu’un bloc de code, par exemple, un pool de connexions qui existe pendant toute la durée de vie de l’application. Dans ce cas, gérez-le dans la couche de démarrage et d’arrêt de l’application.
• La logique n’a pas de paire « ouvrir/fermer » clairement définie. Une fonction ordinaire serait plus lisible.
• Vous voulez ignorer toutes les exceptions avec suppress(Exception). Cela cache les bugs exactement comme except: pass.

Remarque : Ne comptez pas sur le garbage collector ou __del__ pour fermer les ressources. Le moment où ils s’exécutent n’est pas garanti, en particulier sur les implémentations comme PyPy. Utilisez with.

10. Conclusion

Les context managers encapsulent le setup et le teardown au même endroit, et garantissent que le teardown s’exécute toujours, même en cas d’exception. Vous avez appris :

• Le protocole __enter__ et __exit__, et la signification de la valeur retournée par __exit__.
• Comment écrire de façon concise avec @contextmanager, avec try/finally autour de yield obligatoire.
• Les utilitaires suppress, ExitStack, closing, nullcontext, ainsi que leurs versions async.
• Comment tester à la fois la branche commit et la branche rollback avec pytest.

Chaque fois que votre code a une paire « acquérir-puis-libérer » (fichiers, verrous, transactions, connexions), envisagez de l’encapsuler dans un context manager.