Context manager e a declaração with em Python

1. Visão geral

Abrir arquivo, conectar ao banco de dados ou manter um lock precisam de uma etapa de limpeza ao final. Esse passo deve ser executado mesmo quando o código no meio lança uma exceção. Python resolve isso com context managers e a declaração with.

Este artigo apresenta como with funciona por baixo dos panos e como escrever seus próprios context managers com class ou com contextlib. Você também aprenderá a tratar exceções em __exit__, como escrever testes para context managers e erros comuns ao usá-los.

2. O problema: limpeza de recursos com try/finally

Veja como ler um arquivo sem usar with:

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

Essa forma está correta mas é verbosa. Também é fácil se esquecer. Se você colocar open() dentro de try, e open() falhar, então f ainda não foi atribuído e o bloco finally vai lançar NameError. A declaração with encapsula todo o padrão acima:

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

print(f.closed)
order_id,customer,amount

True

Ao sair do bloco with, o arquivo foi fechado. Isso é verdade tanto para saída normal quanto para exceções.

3. O protocolo context manager

Qualquer objeto com dois métodos __enter__ e __exit__ pode ser usado com with. Conceitualmente, a declaração:

with EXPR as target:
    BODY

é equivalente ao código a seguir:

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)

Há três pontos a lembrar:

  • O valor atribuído a target é o resultado de __enter__(), não a própria EXPR.
  • __exit__ é sempre chamado. Quando há uma exceção, ele recebe (exc_type, exc_value, traceback). Quando não há erros, todos os três parâmetros são None.
  • Se __exit__ retorna um valor truthy, a exceção é suprimida. Caso contrário, a exceção é relançada.

3.1. Escrevendo um context manager com class

O primeiro exemplo é um Timer que mede o tempo de execução de um bloco de código:

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("calcular_soma_pedidos") as t:
    total = sum(range(5_000_000))

print(t.elapsed > 0)
[calcular_soma_pedidos] 0.081s
True

Explicação passo a passo:

  • __enter__ registra o momento inicial e retorna self. Por isso, t é o objeto Timer.
  • __exit__ calcula o tempo, imprime na tela e retorna False. Qualquer exceção não será suprimida.
  • A variável t ainda pode ser usada após o bloco with. A razão é que with não cria um novo escopo.

3.2. Tratando exceções em __exit__

Context managers têm maior impacto quando você precisa reagir de forma diferente dependendo de o bloco de código ter sucesso ou falhado. O exemplo a seguir é uma transação simples em um livro-razão (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)} registros")
        else:
            print(f"rollback por {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("saldo insuficiente")
except ValueError:
    pass

print(len(ledger))
commit 2 registros
rollback por ValueError: saldo insuficiente
2

No segundo bloco with, o pedido número 3 não foi registrado no ledger. Como __exit__ retorna False, o ValueError ainda é lançado e quem chama decide como lidar com ele.

Nota: Retorne True de __exit__ apenas quando você intencionalmente quer suprimir a exceção. Um método __exit__ que acidentalmente faz return self ou retorna outro objeto truthy fará o erro desaparecer sem deixar rastros.

4. Mais conciso com @contextmanager

Escrever uma classe inteira apenas para uma etapa de setup e teardown é bastante verboso. O módulo contextlib fornece o decorator @contextmanager para transformar um gerador em context manager:

from contextlib import contextmanager


class PaymentService:
    def connect(self):
        print("abrindo conexão com o gateway de pagamento")

    def charge(self, customer: str, amount: int):
        print(f"debitando {amount} de {customer}")

    def disconnect(self):
        print("fechando conexão")


@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)
abrindo conexão com o gateway de pagamento
debitando 300000 de Lan
fechando conexão

Como o decorator mapeia o gerador para o protocolo context manager:

  • O código antes de yield desempenha o papel de __enter__.
  • O valor que é yield é o valor atribuído à variável após as.
  • O código após yield desempenha o papel de __exit__. Se o bloco with lançar uma exceção, essa exceção é relançada na posição de yield.

Como a exceção aparece exatamente em yield, o bloco try/finally ao redor de yield é obrigatório. Sem ele, disconnect() não será executado quando houver um erro.

Para tratar cada tipo de erro separadamente, você adiciona except:

@contextmanager
def safe_charge(order_id: int):
    try:
        yield
    except ConnectionError as e:
        print(f"pedido {order_id}: erro de rede, tentarei novamente mais tarde ({e})")


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

print("continuando o processamento")
pedido 42: erro de rede, tentarei novamente mais tarde (timeout)
continuando o processamento

O gerador pega o erro mas não o relança, então a exceção é suprimida. Esse resultado é equivalente a __exit__ retornar True.

5. Utilitários disponíveis no contextlib

5.1. suppress: ignorar um tipo específico de exceção

import os
from contextlib import suppress

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

print("pronto")
pronto

O código acima é mais conciso e deixa clara a intenção em comparação com try: ... except FileNotFoundError: pass.

5.2. ExitStack: número desconhecido de recursos

Suponha que você precise mesclar vários arquivos de fatura, mas o número de arquivos só é conhecido em tempo de execução:

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"fatura {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"))
fatura 0
fatura 1
fatura 2

enter_context chama __enter__ e memoriza o __exit__ correspondente. Quando o ExitStack termina, os arquivos são fechados em ordem inversa de quando foram abertos. Se o terceiro arquivo não abrir, os dois primeiros ainda serão fechados corretamente.

5.3. closing e nullcontext

  • closing(obj) transforma um objeto que tem apenas um método close() em context manager.
  • nullcontext(value) é um context manager que não faz nada. É útil quando um parâmetro pode ou não ser um 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. Múltiplos context managers em uma declaração with

Você pode declarar múltiplos context managers na mesma declaração with. A partir do Python 3.10, a sintaxe com parênteses permite quebras de linha de forma elegante:

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

Essa forma é equivalente a dois blocos with aninhados. src é aberto primeiro e fechado por último.

7. Context manager assíncrono

Para código asyncio, o protocolo correspondente é __aenter__ e __aexit__, usado com async with. A versão com gerador é @asynccontextmanager:

import asyncio
from contextlib import asynccontextmanager


@asynccontextmanager
async def order_queue(name: str):
    print(f"conectando à fila {name}")
    await asyncio.sleep(0.01)
    try:
        yield [f"pedido-{i}" for i in range(3)]
    finally:
        await asyncio.sleep(0.01)
        print(f"fechando fila {name}")


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

asyncio.run(main())
conectando à fila payments
processando pedido-0
processando pedido-1
processando pedido-2
fechando fila payments

Nota: Usar with normal com um async context manager causará TypeError (ou AttributeError em versões antigas do Python). O oposto também é verdadeiro. Esses dois protocolos são separados.

8. Testando context managers

Context managers devem ser testados tanto na branch de sucesso quanto na de erro. Primeiro instale pytest:

pip install pytest

Coloque a classe Transaction da seção 3.2 em um arquivo transaction.py, então escreva os testes:

# 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="saldo"):
        with Transaction(ledger) as tx:
            tx.append({"order_id": 2, "amount": 999})
            raise ValueError("saldo insuficiente")

    assert ledger == []
pytest -q
..                                                       [100%]
2 passed in 0.01s

O segundo teste verifica duas coisas ao mesmo tempo: os dados foram revertidos e a exceção não foi suprimida. O próprio pytest.raises também é um context manager, um exemplo familiar do próprio tema deste artigo.

9. Erros comuns e quando não usar

Esquecer try/finally ao redor de yield. Esse é o erro mais comum com @contextmanager:

@contextmanager
def broken_session(service):
    service.connect()
    yield service
    service.disconnect()  # não é executado se o bloco with lançar uma exceção

Usar um recurso após with terminar. A variável ainda existe, mas o recurso foi liberado:

with open("orders.csv", encoding="utf-8") as f:
    pass
f.read()
ValueError: I/O operation on closed file.

yield múltiplas vezes no gerador. Um gerador usado com @contextmanager deve fazer yield exatamente uma vez. Se fizer yield mais de uma vez, Python lançará RuntimeError: generator didn't stop.

Reutilizar context manager criado de @contextmanager. Esse objeto pode ser usado apenas uma vez. Chamar with uma segunda vez no mesmo objeto falhará. Cada vez que usar, chame a função novamente para criar um novo objeto.

Abuso. Não use context managers nos seguintes casos:

  • Quando o recurso precisa existir por mais tempo que um bloco de código, por exemplo, um pool de conexões que existe durante todo o ciclo de vida da aplicação. Nesse caso, gerencie-o no nível de inicialização e encerramento da aplicação.
  • Quando a lógica não tem um par “abrir/fechar” bem definido. Uma função comum seria mais legível.
  • Quando você quer ignorar todas as exceções com suppress(Exception). Essa abordagem oculta bugs assim como except: pass.

Nota: Não confie em garbage collector ou __del__ para fechar recursos. O momento em que eles são executados não é garantido, especialmente em implementações como PyPy. Use with.

10. Conclusão

Context managers encapsulam setup e teardown em um só lugar e garantem que teardown sempre execute mesmo com exceções. Você viu:

  • O protocolo __enter__ e __exit__, junto com o significado do valor retornado de __exit__.
  • Como escrever de forma concisa com @contextmanager, com try/finally ao redor de yield sendo obrigatório.
  • Os utilitários suppress, ExitStack, closing, nullcontext, junto com a versão async.
  • Como testar ambas as branches de commit e rollback com pytest.

Sempre que seu código tiver um par “adquirir e devolver” (arquivo, lock, transação, conexão), considere encapsulá-lo em um context manager.

pythoncontext-managercontextlibresource-management