Narcis Miclaus

8 min de citit

Python, de la zero Lecția 6 / 60

Context managers si with-statement, dincolo de deschiderea fisierelor

Protocolul din spatele lui `with`, trei moduri de a-ti scrie propriul context manager si cand context managers bat try/finally.

Fiecare începător în Python învață context managers la fel. Ziua unu: deschide un fișier cu open(...). Ziua doi: cineva spune „folosește with open(...) as f: în loc, închide fișierul automat”. Ăsta e tot modelul mental pe care majoritatea oamenilor îl plimbă cu ei ani la rând.

E un punct de plecare bun. E și cam 5% din ce face with. Același protocol gestionează tranzacții de baze de date, achiziție de lock-uri, directoare temporare care se curăță singure, redirecționare de stdout, blocuri de timing, suprimare de excepții și zeci de pattern-uri în care trebuie să faci X la intrare și Y la ieșire, garantat, chiar dacă corpul blocului ridică o excepție.

Astăzi ne uităm la protocolul de dedesubt, la trei moduri de a-ți scrie propriul context manager și la cazurile în care with e dramatic mai plăcut decât try/finally.

Protocolul

Un context manager e orice obiect cu două metode:

  • __enter__(self): chemat când execuția intră în blocul with. Ce returnează e legat la variabila de după as.
  • __exit__(self, exc_type, exc_value, traceback): chemat când execuția părăsește blocul, fie normal, fie via o excepție.

Ăsta e tot protocolul. PEP 343 l-a oficializat; tot restul sunt biblioteci construite peste.

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

Chiar dacă corpul a ridicat o excepție, __exit__ a rulat. Asta e garanția pe care ți-o dă with. E aceeași garanție pe care ți-o dă try/finally, dar cu mult mai puțin cod pe pagină și un semnal mai clar al intenției.

Valoarea de retur din __exit__ (nu te-ncurca aici)

__exit__ poate returna o valoare. Dacă returnează o valoare truthy, excepția e suprimată: execuția continuă după blocul with ca și cum nu s-ar fi întâmplat nimic. Dacă returnează None sau False, excepția se propagă normal.

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

    def __exit__(self, exc_type, exc_value, traceback) -> bool:
        return True   # suprimă toate excepțiile


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

print("we get here")

În 99% din cazuri, nu vrei asta. Înghițirea tăcută a excepțiilor ascunde bug-uri. Utilizările legitime sunt înguste: contextlib.suppress, handler-e de rollback de tranzacții care reraise după curățare, decoratori de retry construiți ca context managers. Dacă te trezești scriind return True dintr-un __exit__, oprește-te și întreabă-te dacă chiar asta vrei. De obicei, întoarcerea lui None (default-ul) e corectă.

Modul 1: o clasă cu __enter__ / __exit__

Forma cea mai explicită. Cea mai bună atunci când managerul are stare care trăiește între enter/exit, sau când utilizatorul are nevoie să cheme metode pe el în interiorul blocului.

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
    # ... fă treabă ...

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

Instanța Timer supraviețuiește dincolo de blocul with: t.elapsed_ms e încă citibil. Ăsta e cazul pentru forma de clasă: când ai nevoie ca managerul să fie un obiect real cu care interacționezi după.

Modul 2: @contextlib.contextmanager pe un generator

Pentru cazul comun în care __enter__ e scurt și __exit__ e scurt, forma de clasă e exagerată. contextlib îți dă un decorator care transformă o funcție generator într-un context manager. Tot ce e înainte de yield e __enter__. Tot ce e după yield e __exit__. Valoarea livrată cu yield e ce leagă as.

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"):
    # ... lucrează în /tmp ...
    pass
# înapoi în directorul original, chiar dacă blocul a ridicat o excepție

try/finally e obligatoriu aici. Dacă corpul ridică o excepție, controlul se întoarce la generator, la locul yield, ca o excepție: fără finally, os.chdir(previous) ar fi sărit, iar tu ai scurge schimbarea de director de lucru. Ăsta e cel mai comun bug în context managers scriși de mână; pytest îl prinde prima dată când ceva aruncă.

O tranzacție de bază de date e exemplul de manual:

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 la succes, ROLLBACK la eșec. Cursorul mereu închis.

Ăsta e pattern-ul pe care îl expun majoritatea ORM-urilor și driver-elor de DB sub nume de tipul session.begin() sau engine.begin(). Oasele sunt mereu aceleași.

Modul 3: contextlib.ExitStack pentru contexte dinamice

Uneori nu știi la write-time câte contexte vei avea nevoie. Deschizi un fișier per rând dintr-o configurare. Sau N conexiuni la baza de date pe baza unor argumente de runtime. Sau un număr variabil de lock-uri. Nu poți scrie with a, b, c: pentru că nu știi câți a, b, c sunt.

ExitStack e răspunsul:

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)
    # Toate fișierele închise aici, în ordine inversă, chiar și la excepție.

stack.enter_context(cm) e echivalent cu a intra în cm și a-i înregistra __exit__ să fie chemat când stack-ul se desface. Poți înregistra și callback-uri arbitrare:

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)

        # ... lucrează în `workspace` ...
    # workspace e curățat chiar dacă blocul a ridicat o excepție

ExitStack e și unealta corectă atunci când vrei unele curățări condiționate de ce s-a întâmplat mai devreme în bloc: poți folosi pop_all() pentru a detașa curățările înregistrate și a le rula mai târziu (sau niciodată).

Built-ins utile din contextlib

from contextlib import suppress, redirect_stdout, nullcontext
import io


# 1. suppress: ignoră un tip specific de excepție
with suppress(FileNotFoundError):
    os.remove("maybe-doesnt-exist.txt")

# 2. redirect_stdout: capturează output-ul de la print()
buf = io.StringIO()
with redirect_stdout(buf):
    print("captured")
print(buf.getvalue())   # "captured\n"

# 3. nullcontext: un context manager fără efect - util când vrei
#    condițional să intri într-un context real sau în niciunul
def maybe_lock(use_lock: bool, lock):
    cm = lock if use_lock else nullcontext()
    with cm:
        do_critical_work()

suppress e cazul de utilizare legitim pentru un __exit__ care returnează True: pentru un singur tip de excepție specific, numit. Nu îl folosi pentru „ignoră tot”; folosește-l pentru „această excepție exactă e așteptată și înseamnă că operația e un no-op”.

Exemple din lumea reală pe care le-am folosit

Un wrapper „log și reraise” pentru granița unui serviciu:

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 temporar de feature flag pentru testare:

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 cu domeniu, care înregistrează cine îl deține:

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()

(Notă laterală: acel global e o capcană în cod concurent real; e ok pentru diagnosticare, nu pentru corectitudine.)

Async context managers (preview)

asyncio are propria versiune. Protocolul e __aenter__ / __aexit__, ambele async, iar sintaxa e async with:

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

Aceeași idee, async până jos. Vom acoperi-o cum trebuie în lecția 40, când ajungem la I/O async: pattern-urile se transferă aproape verbatim.

Când with bate try/finally

Ambele sunt corecte. Motivele de a prefera with:

  • Intenție. with transaction(conn): spune exact ce se întâmplă. try: ... finally: spune „ceva are nevoie de curățare, derulează în jos ca să afli ce”.
  • Reutilizare. Un context manager pe care-l scrii o dată e reutilizabil în douăzeci de locuri. Un bloc try/finally e copy-paste-uit în douăzeci de locuri.
  • Compoziție. with a, b, c: intră în trei manageri în ordine, iese din ei în ordine inversă. ExitStack îți permite să faci același lucru dinamic. try/finally nu se compune: ai face nesting, iar ordinea curățărilor devine un puzzle.
  • __exit__ rulează mereu. E greu să sari accidental peste curățare cu with. Cu try/finally poți uita finally-ul, sau poți pune codul greșit în afara lui.

try/finally e încă alegerea corectă atunci când curățarea e unică pentru un singur loc și nu merită un nume. Nu face un context manager numit _temp_close_thing care e folosit într-o singură funcție.

Asta a fost despre context managers: instrucțiunea with, protocolul, trei moduri de a-ți scrie propriul context manager și câteva pattern-uri care apar în producție. Modulul următor schimbă viteza: data classes, typing și sistemul modern de tipuri din Python care face ca tot codul pe care l-am scris să fie chiar safe de refactorizat.

Citations (consultat 2026-05-01):

Caută