Python, dalle fondamenta Lezione 6 / 60

Context manager e l'istruzione with, oltre l'apertura dei file

Il protocollo dietro `with`, tre modi per scriverne uno tuo, e quando i context manager battono try/finally.

Ogni principiante Python impara i context manager nello stesso modo. Giorno uno: apri un file con open(...). Giorno due: qualcuno gli dice “usa with open(...) as f: invece, chiude il file da solo.” Quello è l’intero modello mentale che la maggior parte delle persone si porta dietro per anni.

È un buon punto di partenza. È anche circa il 5% di quello che with fa. Lo stesso protocollo gestisce le transazioni di database, l’acquisizione di lock, le directory temporanee che si ripuliscono da sole, lo stdout reindirizzato, i blocchi di timing, la soppressione di eccezioni, e decine di pattern in cui devi fare X all’ingresso e Y all’uscita, garantito, anche se il corpo del blocco solleva un’eccezione.

Oggi guardiamo il protocollo che sta sotto, tre modi per scriverne uno tuo, e i casi in cui with è drammaticamente più piacevole di try/finally.

Il protocollo

Un context manager è qualunque oggetto con due metodi:

  • __enter__(self) viene chiamato quando l’esecuzione entra nel blocco with. Quello che restituisce viene legato alla variabile dopo as.
  • __exit__(self, exc_type, exc_value, traceback) viene chiamato quando l’esecuzione esce dal blocco, sia normalmente sia per via di un’eccezione.

È tutto qui il protocollo. La PEP 343 lo ha reso ufficiale; tutto il resto è librerie costruite sopra.

class Sandbox:
    def __enter__(self) -> "Sandbox":
        print("entering")
        return self

    def __exit__(self, exc_type, exc_value, traceback) -> None:
        print(f"exiting (exception: {exc_type})")


with Sandbox() as box:
    print("inside")
    raise ValueError("oops")
# entering
# inside
# exiting (exception: <class 'ValueError'>)
# Traceback (most recent call last):
#   ...
# ValueError: oops

Anche se il corpo ha sollevato, __exit__ è girato. Quella è la garanzia che with ti dà. È la stessa garanzia che ti dà try/finally, ma con molto meno codice sulla pagina e un segnale di intento più chiaro.

Il valore di ritorno di __exit__ (non sbagliarlo)

__exit__ può restituire un valore. Se restituisce un valore truthy, l’eccezione viene soppressa: l’esecuzione prosegue dopo il blocco with come se non fosse successo niente. Se restituisce None o False, l’eccezione si propaga normalmente.

class SwallowEverything:
    def __enter__(self) -> "SwallowEverything":
        return self

    def __exit__(self, exc_type, exc_value, traceback) -> bool:
        return True   # sopprime tutte le eccezioni


with SwallowEverything():
    raise ValueError("ignored silently")

print("we get here")

Il 99% delle volte non vuoi questo. Inghiottire le eccezioni in silenzio nasconde i bug. Gli usi legittimi sono pochi: contextlib.suppress, gli handler di rollback delle transazioni che ri-sollevano dopo la pulizia, i decorator di retry costruiti come context manager. Se ti ritrovi a scrivere return True da un __exit__, fermati e chiediti se è davvero quello che vuoi. Di solito, restituire None (il default) è la scelta giusta.

Modo 1: una classe con __enter__ / __exit__

La forma più esplicita. Va meglio quando il manager ha stato che vive attraverso enter/exit, o quando l’utente ha bisogno di chiamare metodi su di esso dentro il blocco.

import time
from typing import Optional


class Timer:
    def __init__(self, label: str) -> None:
        self.label = label
        self.elapsed_ms: float = 0.0
        self._start: Optional[float] = None

    def __enter__(self) -> "Timer":
        self._start = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc_value, traceback) -> None:
        assert self._start is not None
        self.elapsed_ms = (time.perf_counter() - self._start) * 1000
        print(f"{self.label}: {self.elapsed_ms:.2f} ms")


with Timer("import csv") as t:
    import csv
    # ... do work ...

print(f"that took {t.elapsed_ms:.2f} ms total")

L’istanza di Timer sopravvive al blocco with: t.elapsed_ms resta leggibile. Quello è il caso d’uso della forma a classe: quando hai bisogno che il manager sia un oggetto vero con cui interagisci dopo.

Modo 2: @contextlib.contextmanager su un generator

Per il caso comune in cui __enter__ è breve e __exit__ è breve, la forma a classe è esagerata. contextlib ti dà un decorator che trasforma una generator function in un context manager. Tutto quello che sta prima di yield è __enter__. Tutto quello che sta dopo yield è __exit__. Il valore restituito da yield è quello a cui as viene legato.

from contextlib import contextmanager
from typing import Iterator
import os


@contextmanager
def in_directory(path: str) -> Iterator[None]:
    previous = os.getcwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(previous)


with in_directory("/tmp"):
    # ... lavoro in /tmp ...
    pass
# torniamo nella directory originale, anche se il blocco ha sollevato

Il try/finally qui è obbligatorio. Se il corpo solleva, il controllo torna al generator nel punto dello yield sotto forma di eccezione: senza finally, l’os.chdir(previous) verrebbe saltato, e ti porteresti dietro il cambio di working directory. È il bug più comune nei context manager scritti a mano; pytest lo intercetta la prima volta che qualcosa solleva.

Una transazione di database è l’esempio da manuale:

from contextlib import contextmanager
from typing import Iterator


@contextmanager
def transaction(conn) -> Iterator[None]:
    cursor = conn.cursor()
    try:
        cursor.execute("BEGIN")
        yield
        cursor.execute("COMMIT")
    except Exception:
        cursor.execute("ROLLBACK")
        raise
    finally:
        cursor.close()


with transaction(conn):
    cursor = conn.cursor()
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
# COMMIT in caso di successo, ROLLBACK in caso di errore. Cursore sempre chiuso.

È il pattern che la maggior parte degli ORM e dei driver DB espone con nomi tipo session.begin() o engine.begin(). L’ossatura è sempre la stessa.

Modo 3: contextlib.ExitStack per context dinamici

A volte non sai al momento della scrittura quanti context ti serviranno. Stai aprendo un file per ogni riga di una config. O N connessioni a database basate su argomenti runtime. O un numero variabile di lock. Non puoi scrivere with a, b, c: perché non sai quanti a, b, c ci sono.

ExitStack è la risposta:

from contextlib import ExitStack
from typing import IO


def merge_files(paths: list[str], output: str) -> None:
    with ExitStack() as stack:
        inputs: list[IO[str]] = [
            stack.enter_context(open(p, encoding="utf-8"))
            for p in paths
        ]
        out = stack.enter_context(open(output, "w", encoding="utf-8"))

        for f in inputs:
            for line in f:
                out.write(line)
    # Tutti i file vengono chiusi qui, in ordine inverso, anche su eccezione.

stack.enter_context(cm) equivale a entrare in cm e a registrare il suo __exit__ perché venga chiamato quando lo stack si srotola. Puoi registrare anche callback arbitrari:

import shutil
import tempfile
from contextlib import ExitStack


def process_with_workspace(input_path: str) -> None:
    with ExitStack() as stack:
        workspace = tempfile.mkdtemp()
        stack.callback(shutil.rmtree, workspace, ignore_errors=True)

        # ... lavoro in `workspace` ...
    # workspace viene ripulito anche se il corpo ha sollevato

ExitStack è anche lo strumento giusto quando vuoi che qualche pulizia sia condizionale a quello che è successo prima nel blocco: puoi chiamare pop_all() per staccare le pulizie registrate ed eseguirle dopo (o mai).

Built-in utili da contextlib

from contextlib import suppress, redirect_stdout, nullcontext
import io


# 1. suppress: ignora un tipo specifico di eccezione
with suppress(FileNotFoundError):
    os.remove("maybe-doesnt-exist.txt")

# 2. redirect_stdout: cattura l'output di print()
buf = io.StringIO()
with redirect_stdout(buf):
    print("captured")
print(buf.getvalue())   # "captured\n"

# 3. nullcontext: un context manager no-op, utile quando vuoi
#    condizionalmente entrare in un context vero o in nessuno
def maybe_lock(use_lock: bool, lock):
    cm = lock if use_lock else nullcontext()
    with cm:
        do_critical_work()

suppress è il caso d’uso legittimo per un __exit__ che restituisce True, ma per un singolo, specifico tipo di eccezione con un nome. Non usarlo per “ignora tutto”; usalo per “questa esatta eccezione è attesa e significa che l’operazione è un no-op.”

Esempi reali che ho usato

Un wrapper “log e ri-solleva” per il bordo di un servizio:

from contextlib import contextmanager
from typing import Iterator
import logging
import time


@contextmanager
def request_context(name: str, logger: logging.Logger) -> Iterator[None]:
    start = time.perf_counter()
    logger.info("start", extra={"op": name})
    try:
        yield
    except Exception:
        elapsed = (time.perf_counter() - start) * 1000
        logger.exception("failed", extra={"op": name, "ms": elapsed})
        raise
    else:
        elapsed = (time.perf_counter() - start) * 1000
        logger.info("ok", extra={"op": name, "ms": elapsed})


with request_context("export-orders", log):
    run_export()

Un override temporaneo di un feature flag per i test:

from contextlib import contextmanager
from typing import Iterator


@contextmanager
def feature(flag: str, value: bool) -> Iterator[None]:
    previous = flags.get(flag)
    flags[flag] = value
    try:
        yield
    finally:
        if previous is None:
            del flags[flag]
        else:
            flags[flag] = previous


def test_new_export() -> None:
    with feature("new_export_pipeline", True):
        result = run_export()
    assert result.path.endswith(".parquet")

Un lock con scope che registra chi lo ha tenuto:

from contextlib import contextmanager
from threading import RLock
from typing import Iterator


_lock = RLock()
_holder: str | None = None


@contextmanager
def held_by(name: str) -> Iterator[None]:
    global _holder
    _lock.acquire()
    _holder = name
    try:
        yield
    finally:
        _holder = None
        _lock.release()

(Nota a margine: quel global è un pericolo in codice davvero concorrente; va bene per la diagnostica, non per la correttezza.)

Context manager async (anteprima)

asyncio ha la sua versione. Il protocollo è __aenter__ / __aexit__, entrambi async, e la sintassi è async with:

async with session.get(url) as response:
    body = await response.read()

Stessa idea, async fino in fondo. Lo copriremo per bene nella lezione 40 quando arriviamo all’async I/O: i pattern si trasferiscono quasi parola per parola.

Quando with batte try/finally

Sono entrambi corretti. I motivi per preferire with:

  • Intento. with transaction(conn): dice esattamente cosa sta succedendo. try: ... finally: dice “qualcosa ha bisogno di pulizia, scorri giù per scoprire cosa.”
  • Riuso. Un context manager scritto una volta è riutilizzabile in venti posti. Un blocco try/finally viene copia-incollato in venti posti.
  • Composizione. with a, b, c: entra in tre manager in ordine, esce in ordine inverso. ExitStack ti permette di farlo dinamicamente. try/finally non si compone: dovresti annidare, e l’ordine di pulizia diventa un puzzle.
  • L’__exit__ gira sempre. È difficile saltare la pulizia per sbaglio con with. Con try/finally puoi dimenticare il finally, o mettere il codice sbagliato fuori da esso.

try/finally resta la scelta giusta quando la pulizia è unica per un solo posto e non merita un nome. Non fare un context manager che si chiama _temp_close_thing e che viene usato in esattamente una funzione.

Questi sono i context manager: l’istruzione with, il protocollo, tre modi per scriverne uno tuo, e qualche pattern che salta fuori in produzione. La prossima lezione cambia marcia: data class, typing, e il sistema di tipi del Python moderno che rende davvero sicuro fare refactoring sul codice che abbiamo scritto finora.


Citations (retrieved 2026-05-01):

Cerca