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ópriaEXPR. __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ãoNone.- 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 retornaself. Por isso,té o objetoTimer.__exit__calcula o tempo, imprime na tela e retornaFalse. Qualquer exceção não será suprimida.- A variável
tainda pode ser usada após o blocowith. A razão é quewithnã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
Truede__exit__apenas quando você intencionalmente quer suprimir a exceção. Um método__exit__que acidentalmente fazreturn selfou 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
yielddesempenha o papel de__enter__. - O valor que é
yieldé o valor atribuído à variável apósas. - O código após
yielddesempenha o papel de__exit__. Se o blocowithlançar uma exceção, essa exceção é relançada na posição deyield.
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étodoclose()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
withnormal com um async context manager causaráTypeError(ouAttributeErrorem 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 comoexcept: 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. Usewith.
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, comtry/finallyao redor deyieldsendo 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.