contextlib: ExitStack, suppress y otros utilitarios

El módulo contextlib resuelve tres problemas concretos que aparecen cuando trabajas con context managers en código real: gestionar un número de recursos que no conoces hasta el momento de ejecución, ignorar excepciones específicas sin escribir bloques try/except vacíos, y adaptar objetos que tienen .close() pero no implementan el protocolo de context manager.

El núcleo de cada utilidad

ExitStack es un context manager que actúa como contenedor de otros context managers. Su propósito es diferir la decisión de cuántos y cuáles recursos abrir hasta tiempo de ejecución. Internamente mantiene una pila de callbacks de limpieza; cuando el bloque with termina (con o sin excepción), ejecuta esos callbacks en orden LIFO, exactamente como lo haría un anidamiento estático de bloques with. Úsalo cuando el número de recursos depende de datos externos, como una lista de archivos recibida por parámetro. Si abres diez archivos en un bucle sin ExitStack, un error en el quinto te deja los cuatro anteriores sin cerrar a menos que escribas lógica de cleanup manual.

contextlib.suppress(*exceptions) devuelve un context manager que atrapa y silencia las excepciones indicadas. Es la forma idiomática de expresar “si falla con este error específico, ignóralo y sigue”. La alternativa, un bloque try/except: pass, oculta la intención y es fácil de ampliar accidentalmente. Úsalo cuando la excepción es un caso esperado y normal, no un error: borrar un fichero que quizás no existe, acceder a una clave de un diccionario cuando su ausencia es un estado válido.

contextlib.nullcontext(enter_result=None) es un context manager que no hace absolutamente nada. Existe para eliminar bifurcaciones condicionales del estilo if lock: with lock: ... vs else: .... Cuando una función acepta un context manager opcional, puedes pasar nullcontext() como valor por defecto y el código del cuerpo queda limpio.

contextlib.closing(thing) adapta cualquier objeto con método .close() al protocolo de context manager. Muchos objetos de bibliotecas heredadas (conexiones de red, handles de bases de datos, generadores) tienen .close() pero no implementan __enter__/__exit__. closing envuelve eso en un with garantizando la llamada a .close() aunque ocurra una excepción.

import contextlib
import os
import threading
from pathlib import Path


def process_reports(paths: list[str], lock: threading.Lock | None = None) -> dict[str, int]:
    """
    Lee varios archivos de texto y devuelve un mapa nombre → número de líneas.
    `lock` es opcional; si no se pasa, no se usa ningún lock.
    """
    results: dict[str, int] = {}

    # nullcontext como valor por defecto evita un if/else alrededor del with
    sync = lock if lock is not None else contextlib.nullcontext()

    with contextlib.ExitStack() as stack:
        # Abrimos cada archivo y lo registramos en la pila de limpieza.
        # Si open() falla en el archivo N, los N-1 anteriores ya están
        # registrados y ExitStack los cerrará al salir del bloque.
        file_handles = [
            stack.enter_context(open(p, encoding="utf-8"))
            for p in paths
        ]

        # El lock (real o nullcontext) también entra en la pila
        stack.enter_context(sync)

        for handle in file_handles:
            line_count = sum(1 for _ in handle)
            results[handle.name] = line_count

    return results


def cleanup_temp_files(temp_dir: str) -> None:
    """Borra archivos .tmp; ignora silenciosamente los que ya no existen."""
    tmp_path = Path(temp_dir)

    for item in tmp_path.glob("*.tmp"):
        # suppress es más expresivo que try/except FileNotFoundError: pass
        # porque declara la intención: este error es un caso normal, no un bug
        with contextlib.suppress(FileNotFoundError, PermissionError):
            item.unlink()


class LegacyConnection:
    """Simula una conexión de biblioteca heredada: tiene .close() pero no with."""

    def __init__(self, host: str):
        self.host = host
        self._open = True
        print(f"Conexión abierta a {host}")

    def query(self, sql: str) -> list[str]:
        if not self._open:
            raise RuntimeError("Conexión ya cerrada")
        return [f"row_{i}" for i in range(3)]

    def close(self) -> None:
        self._open = False
        print(f"Conexión cerrada a {self.host}")


def run_legacy_query(host: str, sql: str) -> list[str]:
    # closing() garantiza conn.close() sin modificar LegacyConnection
    with contextlib.closing(LegacyConnection(host)) as conn:
        return conn.query(sql)


if __name__ == "__main__":
    # --- ExitStack + nullcontext ---
    # Creamos archivos de prueba temporales
    test_files = []
    for i in range(3):
        p = f"/tmp/report_{i}.txt"
        Path(p).write_text("línea1\nlínea2\nlínea3\n", encoding="utf-8")
        test_files.append(p)

    counts = process_reports(test_files)
    print(counts)  # {'/tmp/report_0.txt': 3, ...}

    # Con lock real
    lock = threading.Lock()
    counts_locked = process_reports(test_files, lock=lock)
    print(counts_locked)

    # --- suppress ---
    Path("/tmp/test_suppress_dir").mkdir(exist_ok=True)
    Path("/tmp/test_suppress_dir/old.tmp").touch()
    cleanup_temp_files("/tmp/test_suppress_dir")

    # --- closing ---
    rows = run_legacy_query("db.internal", "SELECT 1")
    print(rows)

Qué está pasando en cada pieza

process_reports es el ejemplo donde ExitStack brilla. La lista paths llega en tiempo de ejecución; no podemos escribir with open(a) as f1, open(b) as f2 cuando no sabemos cuántos elementos hay. La comprensión de lista llama a stack.enter_context(open(p)) por cada ruta: enter_context ejecuta __enter__ del objeto dado y registra su __exit__ en la pila interna. Si el tercer open() lanza FileNotFoundError, ExitStack.__exit__ itera la pila en reversa cerrando los dos archivos ya abiertos antes de propagar la excepción. Es exactamente el invariante que querrías mantener manualmente, pero automático.

La parte con nullcontext muestra el patrón “parámetro de context manager opcional”. Sin nullcontext, escribirías:

if lock:
    with lock:
        # código
else:
    # código duplicado o función auxiliar

Con nullcontext, el cuerpo del with existe una sola vez. El context manager nulo simplemente devuelve None en __enter__ y no hace nada en __exit__. El coste es literalmente cero; no hay overhead real.

En cleanup_temp_files, fíjate en que suppress acepta múltiples tipos de excepción en una sola llamada. El efecto es idéntico a except (FileNotFoundError, PermissionError): pass, pero la semántica que comunica es diferente: “este bloque puede fallar de estas formas y eso está bien”. Un try/except: pass genérico enmascara errores inesperados; suppress con tipos concretos es quirúrgico.

closing en run_legacy_query no modifica LegacyConnection en absoluto, lo que es importante cuando trabajas con código de terceros o bibliotecas que no puedes tocar. Internamente closing es trivial: su __exit__ llama self.thing.close(). Podría parecer demasiado simple para existir, pero la alternativa —un try/finally explícito cada vez que usas esa clase— acumula ruido en el código que consume esos objetos.

Errores que debes conocer

Error: pasar suppress fuera del with como decorador o usarlo para silenciar excepciones dentro de un ExitStack esperando que afecte al stack entero — suppress solo cubre su propio bloque with.

# ❌ Wrong
with contextlib.ExitStack() as stack:
    contextlib.suppress(ValueError)  # no hace nada: no está en un with
    stack.enter_context(some_resource())

# ✅ Right
with contextlib.ExitStack() as stack:
    stack.enter_context(contextlib.suppress(ValueError))
    stack.enter_context(some_resource())

Si quieres que suppress viva dentro del stack (por ejemplo, para que cubra el __exit__ de otro context manager registrado), regístralo con enter_context.

Error: usar ExitStack y asumir que las excepciones durante __enter__ no propagan porque “el stack las gestiona”.

# ❌ Wrong: si open() lanza, la excepción sube igualmente
with contextlib.ExitStack() as stack:
    for p in paths:
        stack.enter_context(open(p))  # FileNotFoundError propaga si el archivo no existe

# ✅ Right: decide explícitamente si quieres ignorar archivos faltantes
with contextlib.ExitStack() as stack:
    for p in paths:
        with contextlib.suppress(FileNotFoundError):
            stack.enter_context(open(p))

ExitStack garantiza el cierre de lo que ya se registró; no atrapa errores durante la apertura. Son dos responsabilidades distintas.

123

Dejar un comentario

Tu dirección de correo electrónico no será publicada. Los campos obligatorios están marcados con *

Scroll al inicio