Context Managers and the with Statement in Python
1. Overview
Opening files, establishing database connections, or holding locks all require a cleanup step at the end. This step must run even if the code in between throws an exception. Python solves this with context managers and the with statement.
This article explains how with works under the hood, how to write context managers yourself using classes or contextlib, how to handle exceptions in __exit__, how to test context managers, and common pitfalls when using them.
2. The Problem: Cleaning Up Resources with try/finally
Here’s how you’d read a file without using with:
f = open("orders.csv", encoding="utf-8")
try:
header = f.readline()
print(header)
finally:
f.close()
This approach is correct but verbose. It’s also easy to forget. If you put open() inside the try and open() fails, then f hasn’t been assigned yet and the finally block will raise a NameError. The with statement wraps this entire pattern:
with open("orders.csv", encoding="utf-8") as f:
header = f.readline()
print(header)
print(f.closed)
order_id,customer,amount
True
When exiting the with block, the file has been closed. This is true both when exiting normally and when an exception occurs.
3. The Context Manager Protocol
Any object with __enter__ and __exit__ methods can be used with with. Conceptually, the statement:
with EXPR as target:
BODY
is equivalent to the following code:
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)
Three points to remember:
- The value assigned to
targetis the result of__enter__(), notEXPRitself. __exit__is always called. When there’s an exception, it receives(exc_type, exc_value, traceback). When there’s no error, all three parameters areNone.- If
__exit__returns a truthy value, the exception is suppressed. Otherwise, the exception is re-raised.
3.1. Writing a Context Manager with a Class
The first example is a Timer that measures how long a block of code takes to run:
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("compute_order_sum") as t:
total = sum(range(5_000_000))
print(t.elapsed > 0)
[compute_order_sum] 0.081s
True
Breaking it down:
__enter__records the start time and returnsself. As a result,tis theTimerobject itself.__exit__calculates the elapsed time, prints it, and returnsFalse. Any exception that occurs won’t be suppressed.- The variable
tis still usable after thewithblock. The reason is thatwithdoesn’t create a new scope.
3.2. Handling Exceptions in __exit__
Context managers are most useful when you need different behavior depending on whether the block succeeds or fails. The example below is a simple transaction on a 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)} records")
else:
print(f"rollback due to {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("insufficient balance")
except ValueError:
pass
print(len(ledger))
commit 2 records
rollback due to ValueError: insufficient balance
2
In the second with block, order 3 is not written to the ledger. Since __exit__ returns False, the ValueError is still raised and the caller decides how to handle it.
Note: Only return
Truefrom__exit__when you intentionally want to suppress an exception. An__exit__method that accidentallyreturn selfor returns another truthy object will make errors disappear without a trace.
4. More Concisely with @contextmanager
Writing an entire class just for setup and teardown is quite verbose. The contextlib module provides the @contextmanager decorator to turn a generator into a context manager:
from contextlib import contextmanager
class PaymentService:
def connect(self):
print("opening payment gateway connection")
def charge(self, customer: str, amount: int):
print(f"charging {amount} from {customer}")
def disconnect(self):
print("closing connection")
@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)
opening payment gateway connection
charging 300000 from Lan
closing connection
How the decorator maps the generator to the context manager protocol:
- Code before
yieldacts as__enter__. - The value yielded is what gets assigned to the variable after
as. - Code after
yieldacts as__exit__. If thewithblock raises an exception, it’s re-raised at theyieldstatement.
Because the exception appears right at yield, the try/finally block around yield is mandatory. Without it, disconnect() won’t run when there’s an error.
To handle different error types separately, use except:
@contextmanager
def safe_charge(order_id: int):
try:
yield
except ConnectionError as e:
print(f"order {order_id}: network error, will retry later ({e})")
with safe_charge(42):
raise ConnectionError("timeout")
print("continuing processing")
order 42: network error, will retry later (timeout)
continuing processing
The generator catches the error but doesn’t re-raise it, so the exception is suppressed. This is equivalent to __exit__ returning True.
5. Built-in Utilities in contextlib
5.1. suppress: Ignore a Specific Exception Type
import os
from contextlib import suppress
with suppress(FileNotFoundError):
os.remove("cache_orders.tmp")
print("done")
done
This is shorter and clearer in intent than try: ... except FileNotFoundError: pass.
5.2. ExitStack: Unknown Number of Resources
Suppose you need to merge multiple invoice files, but the number of files is only known at runtime:
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"invoice {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"))
invoice 0
invoice 1
invoice 2
enter_context calls __enter__ and remembers the corresponding __exit__. When ExitStack exits, the files are closed in reverse order of opening. If the third file fails to open, the first two are still closed properly.
5.3. closing and nullcontext
closing(obj)turns an object with only aclose()method into a context manager.nullcontext(value)is a context manager that does nothing. It’s useful when a parameter might or might not be a 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. Multiple Context Managers in One with Statement
You can declare multiple context managers in a single with statement. Starting with Python 3.10, the syntax with parentheses allows clean line breaks:
with (
open("orders.csv", encoding="utf-8") as src,
open("orders_backup.csv", "w", encoding="utf-8") as dst,
):
dst.write(src.read())
This is equivalent to two nested with blocks. src is opened first and closed last.
7. Async Context Managers
For asyncio code, the corresponding protocol is __aenter__ and __aexit__, used with async with. The generator version is @asynccontextmanager:
import asyncio
from contextlib import asynccontextmanager
@asynccontextmanager
async def order_queue(name: str):
print(f"connecting to queue {name}")
await asyncio.sleep(0.01)
try:
yield [f"order-{i}" for i in range(3)]
finally:
await asyncio.sleep(0.01)
print(f"closing queue {name}")
async def main():
async with order_queue("payments") as messages:
for m in messages:
print("processing", m)
asyncio.run(main())
connecting to queue payments
processing order-0
processing order-1
processing order-2
closing queue payments
Note: Using regular
withwith an async context manager will cause aTypeError(orAttributeErroron older Python versions). And vice versa. These two protocols are separate.
8. Testing Context Managers
Context managers should be tested for both success and failure paths. First install pytest:
pip install pytest
Put the Transaction class from section 3.2 in a file called transaction.py, then write tests:
# 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="insufficient"):
with Transaction(ledger) as tx:
tx.append({"order_id": 2, "amount": 999})
raise ValueError("insufficient balance")
assert ledger == []
pytest -q
.. [100%]
2 passed in 0.01s
The second test checks two things at once: the data is rolled back, and the exception is not suppressed. pytest.raises itself is also a context manager, a perfect example of the topic of this article.
9. Common Mistakes and When Not to Use
Forgetting try/finally around yield. This is the most common mistake with @contextmanager:
@contextmanager
def broken_session(service):
service.connect()
yield service
service.disconnect() # won't run if the with block raises an exception
Using a resource after the with block ends. The variable still exists, but the resource has been released:
with open("orders.csv", encoding="utf-8") as f:
pass
f.read()
ValueError: I/O operation on closed file.
Yielding multiple times in a generator. A generator used with @contextmanager must yield exactly once. If it yields more, Python raises RuntimeError: generator didn't stop.
Reusing a context manager created with @contextmanager. This object can only be used once. Calling with a second time on the same object will fail. Each time you use it, call the function again to create a new object.
Overuse. You shouldn’t use context managers in these cases:
- When a resource needs to live longer than a single block, such as a connection pool that exists for the application’s lifetime. In that case, manage it at the startup and shutdown level of your application.
- Logic that doesn’t have a clear “open/close” pair. A regular function would be easier to read.
- When you want to suppress all exceptions with
suppress(Exception). This hides bugs just likeexcept: pass.
Note: Don’t rely on the garbage collector or
__del__to close resources. When they run is not guaranteed, especially on implementations like PyPy. Usewithinstead.
10. Conclusion
Context managers encapsulate setup and teardown in one place and ensure teardown always runs even if there’s an exception. You’ve learned about:
- The
__enter__and__exit__protocol and what the return value from__exit__means. - How to write concisely with
@contextmanager, wheretry/finallyaroundyieldis required. - Utilities like
suppress,ExitStack,closing,nullcontext, and their async versions. - How to test both commit and rollback paths using pytest.
When your code has an “acquire then release” pair (file, lock, transaction, connection), consider wrapping it in a context manager.