The contextlib module provides utilities for creating and combining context managers. They control entry into and exit from a with block, guaranteeing cleanup for files, sockets, locks, transactions, temporary directories, and other resources even when an exception occurs. The module reduces repetitive classes and makes lifecycle and error handling explicit.
A context manager does more than close an object. It can establish temporary state, record metrics, redirect streams, begin and finish transactions, or compose a runtime-defined set of resources. The central rule is that every successful acquisition must have a predictable corresponding release.
The context manager protocol
A context manager implements __enter__() and __exit__(). The value returned by __enter__() is bound to the target after as. __exit__() receives exception information when the block fails.
class Resource:
def __enter__(self):
self.open()
return self
def __exit__(self, exc_type, exc_value, traceback):
self.close()
return False
Returning a true value from __exit__() suppresses the exception. Do that only when the error has genuinely been handled.
contextmanager
The @contextmanager decorator converts a generator function into a context manager. Code before yield runs on entry; cleanup in finally runs on exit.
from contextlib import contextmanager
@contextmanager
def temporary_connection():
connection = open_connection()
try:
yield connection
finally:
connection.close()
with temporary_connection() as connection:
connection.execute()
The generator must yield exactly once. Failing to reach the yield or yielding more than once violates the protocol.
Put cleanup in finally
Without finally, an exception raised by the with body may skip release logic.
@contextmanager
def locked_file(path):
file = open(path, "a+")
acquire_lock(file)
try:
yield file
finally:
release_lock(file)
file.close()
If acquisition can fail partway through, release only resources that were actually acquired.
Exceptions inside the generator
When the with body raises, the exception is injected at the yield point. The generator can log, translate, or handle it.
@contextmanager
def log_failures(logger):
try:
yield
except Exception:
logger.exception("block failed")
raise
Re-raise when the problem was not resolved. Logging and silently continuing can conceal corrupted state.
ContextDecorator
Context managers based on ContextDecorator can also decorate functions.
from contextlib import ContextDecorator
class Timer(ContextDecorator):
def __enter__(self):
self.start = now()
return self
def __exit__(self, *exc):
record_duration(now() - self.start)
return False
@Timer()
def process():
run_task()
The object must support repeated use when the decorated function can be called several times.
closing
closing(object) calls close() on exit. It is useful for legacy objects that have a close method but do not implement the context manager protocol.
from contextlib import closing
with closing(open_legacy_resource()) as resource:
resource.use()
Do not wrap an object that already supports with unless necessary. Its native context manager may perform additional work beyond close().
aclosing
aclosing() is the asynchronous counterpart for objects with aclose(), especially asynchronous generators.
from contextlib import aclosing
async with aclosing(async_stream()) as stream:
async for item in stream:
if item.ready:
break
Cleanup occurs in the same asynchronous context, preserving context variables, exceptions, and task lifecycle.
asynccontextmanager
@asynccontextmanager creates asynchronous context managers from async generators.
from contextlib import asynccontextmanager
@asynccontextmanager
async def api_client():
client = await create_client()
try:
yield client
finally:
await client.aclose()
Use it with async with. Cleanup may await I/O, but still needs timeouts and correct cancellation handling.
Reusable and reentrant managers
Some managers are single-use, some can be reused, and some are reentrant. These properties are different.
A generator-based manager creates a fresh instance each time the decorated function is called, so write with resource(): and do not store one instance for repeated entry.
nullcontext
nullcontext() performs no cleanup and returns an optional value. It simplifies code where a resource may already be open.
from contextlib import nullcontext
context = open(path) if path else nullcontext(existing_stream)
with context as stream:
process(stream)
This avoids duplicating the body of the with statement across two branches.
suppress
suppress(*exceptions) ignores selected exception types.
from contextlib import suppress
with suppress(FileNotFoundError):
path.unlink()
Use it only when the exception represents an expected and acceptable outcome. Never suppress broad Exception without a very narrow reason.
Suppression is not logging
If an error requires auditing, retry, metrics, or a user-visible decision, an explicit try/except is clearer. suppress communicates that the missing effect is acceptable and intentionally silent.
redirect_stdout
redirect_stdout(destination) temporarily replaces sys.stdout.
from contextlib import redirect_stdout
from io import StringIO
buffer = StringIO()
with redirect_stdout(buffer):
function_that_prints()
text = buffer.getvalue()
The change is process-global and affects other threads. Use it mainly in scripts, controlled tests, and single-threaded tools.
redirect_stderr
redirect_stderr() performs the same operation for sys.stderr. It does not capture writes made directly to native file descriptors, independent subprocesses, or logging handlers configured elsewhere.
Use subprocess capture options for child processes.
Temporary chdir
chdir(path) changes the current working directory during the block and restores it afterward.
from contextlib import chdir
with chdir("project"):
run_build()
The current directory is process-global state. Do not use this pattern in multithreaded or concurrent programs that depend on relative paths.
ExitStack
ExitStack composes a dynamic number of context managers and cleanup callbacks.
from contextlib import ExitStack
with ExitStack() as stack:
files = [
stack.enter_context(open(path, encoding="utf-8"))
for path in paths
]
combine(files)
If opening the third file fails, the first two close automatically.
LIFO cleanup order
ExitStack runs callbacks in reverse acquisition order. That matches dependent resources: the most recently acquired resource is released first.
Register cleanup immediately after each successful acquisition so there is no leak window.
callback
stack.callback(function, *args, **kwargs) registers a cleanup call that does not receive exception details.
with ExitStack() as stack:
directory = create_temporary_directory()
stack.callback(remove_tree, directory)
run(directory)
Make callbacks idempotent where possible because cleanup may encounter partial state.
push
push() registers the exit portion of a context manager or a function compatible with __exit__. Unlike callback, it can observe and suppress exceptions.
Use that capability carefully because suppression changes what outer callbacks and callers see.
enter_context
enter_context(cm) calls __enter__() and registers __exit__(). It returns the value that a normal as target would receive.
This makes runtime-defined resource sets straightforward.
pop_all
pop_all() transfers callbacks to another stack without executing them. It supports all-or-nothing acquisition.
stack = ExitStack()
try:
resources = [stack.enter_context(open_item(x)) for x in items]
except Exception:
stack.close()
raise
else:
final_stack = stack.pop_all()
After transfer, the new owner must close the returned stack.
AsyncExitStack
AsyncExitStack combines synchronous context managers, asynchronous managers, and async cleanup callbacks.
from contextlib import AsyncExitStack
async with AsyncExitStack() as stack:
clients = [
await stack.enter_async_context(create_client(url))
for url in urls
]
await query(clients)
It is especially useful when the number of asynchronous connections is dynamic.
push_async_callback
Asynchronous callbacks can await flush, shutdown, or resource release. They also run in reverse order.
Apply deadlines because a stuck async cleanup can block application shutdown.
Transactions
A transaction context manager can commit after normal completion and roll back after an exception.
@contextmanager
def transaction(connection):
try:
yield connection
except Exception:
connection.rollback()
raise
else:
connection.commit()
Do not hide the original failure after a rollback unless the caller has another explicit status channel.
Partial acquisition
When setup has several stages, use an internal ExitStack to register every cleanup as it becomes necessary. Transfer the stack only after all stages succeed.
This pattern replaces complicated boolean flags and nested finally blocks.
Cleanup exceptions
An exception during cleanup can replace or chain with the original error. Use specific exception types, logging, and explicit chaining to preserve diagnostics.
Do not ignore a failed commit, flush, or close operation when it is responsible for durability.
Multiple managers in one with
For a fixed number of resources, one with statement with several managers is simpler.
with open_a() as a, open_b() as b:
use(a, b)
Use ExitStack when count or type is dynamic.
Context managers and sockets
Socket objects already support the context protocol and close on exit.
import socket
with socket.create_connection((host, port), timeout=5) as sock:
sock.sendall(data)
See Python socket for timeouts, framing, and shutdown.
Context managers and email
smtplib.SMTP clients also support with, ensuring orderly session closure. See Python smtplib.
Locks
Locks from threading and multiprocessing act as context managers. The block reduces the chance of forgetting release.
with lock:
update_state()
You must still avoid deadlocks, keep critical sections short, and acquire multiple locks in a consistent order.
Context variables
A context manager can temporarily set a ContextVar and reset its token during exit.
@contextmanager
def request_context(value):
token = request_id.set(value)
try:
yield
finally:
request_id.reset(token)
This is useful for logging and tracing, including asynchronous applications.
Metrics
A manager can measure duration and record success or failure on exit while preserving the exception.
Do not allow an observability backend failure to hide the application’s primary error.
Typing
Use typing.ContextManager, AsyncContextManager, or suitable structural protocols when declaring APIs. Annotate the type yielded by generator-based managers.
A clear signature distinguishes the manager object from the resource it produces.
Testing
Test normal exit, exceptions inside the body, acquisition failure, cleanup failure, repeated use, asynchronous cancellation, and partial acquisition. Verify release order.
Mocks should confirm that close, rollback, and callbacks occur exactly when expected.
Common mistakes
Common failures include forgetting finally, yielding more than once in @contextmanager, suppressing exceptions accidentally, redirecting stdout in a multithreaded process, changing the global working directory concurrently, reusing a single-use manager, registering cleanup too late, and forgetting to close a transferred stack.
Conclusion
contextlib makes resource acquisition and release safer and clearer. Use @contextmanager for simple managers, ExitStack for dynamic resources, asynchronous variants for awaitable cleanup, and nullcontext for optional paths.
Keep ownership explicit and never hide important failures. Consult the official contextlib documentation and the context manager protocol reference.
