contextlib
The contextlib module provides utilities for working with context managers. It lets you create context managers using generators, suppress specific exceptions, manage multiple cleanup operations, and temporarily redirect stdout/stderr. These tools simplify resource management and reduce boilerplate code.
Syntax
import contextlib
# Core functions
from contextlib import contextmanager, suppress, nullcontext, closing
from contextlib import redirect_stdout, redirect_stderr
from contextlib import ExitStack, AsyncExitStack, AbstractContextManager
contextmanager
The contextmanager decorator lets you create a context manager from a generator function instead of writing a full class with __enter__ and __exit__ methods. The function must yield exactly once to provide the value for the as clause.
Signature
@contextlib.contextmanager
def managed_resource(*args, **kwargs):
# Setup code
resource = acquire_resource()
try:
yield resource # This value is bound to 'as' variable
finally:
# Cleanup code runs regardless of exception
release_resource(resource)
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
func | callable | — | Generator function to wrap as context manager |
Examples
from contextlib import contextmanager
import os
@contextmanager
def temporary_file(filename):
"""Create a temp file, yield it, then delete on exit."""
f = open(filename, 'w')
try:
yield f
finally:
f.close()
if os.path.exists(filename):
os.remove(filename)
# Using the context manager
with temporary_file('temp.txt') as f:
f.write('Hello, world!')
# File is automatically closed and deleted here
# Output: None (no errors)
# Can also be used as a decorator
@contextmanager
def debug_section(name):
print(f"Entering: {name}")
yield
print(f"Exiting: {name}")
@debug_section("math")
x = 1 + 1
# Output:
# Entering: math
# Exiting: math
Exception handling in contextmanager
from contextlib import contextmanager
@contextmanager
def catch_and_reraise(name):
try:
yield
except ValueError as e:
print(f"Caught {e} in {name}")
raise # Must re-raise if you want exception to propagate
try:
with catch_and_reraise("processor"):
raise ValueError("bad value")
except ValueError:
print("Exception reached caller")
# Output:
# Caught bad value in processor
# Exception reached caller
suppress
The suppress context manager ignores specified exceptions. When an exception occurs in the with block, execution continues normally after the block ends. This is cleaner than wrapping code in try/except/pass.
Signature
contextlib.suppress(*exceptions)
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
*exceptions | Exception | — | Exception types to suppress |
Examples
import os
from contextlib import suppress
# Clean way to remove a file if it exists
with suppress(FileNotFoundError):
os.remove('cache.tmp')
# Suppress multiple exception types
data = {'key': 'value'}
with suppress(KeyError, TypeError, ValueError):
del data['missing_key'] # KeyError suppressed
print("Execution continued past the error")
# Output: Execution continued past the error
nullcontext
The nullcontext context manager does nothing but return a value. It’s useful when you have code that conditionally uses a context manager—instead of writing if/else blocks, you can always use a context manager and swap in nullcontext when you don’t need one.
Signature
contextlib.nullcontext(enter_result=None)
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
enter_result | any | None | Value to return from __enter__ |
Examples
from contextlib import nullcontext
def process_file(filename, mode='r'):
"""Open file if string, use as-is if already a file object."""
if isinstance(filename, str):
cm = open(filename, mode)
else:
# Pass-through: does nothing
cm = nullcontext(filename)
with cm as f:
return f.read()
# Using with a filename
content = process_file('example.txt')
print(content)
# Using with an already-open file
my_file = open('example.txt')
content = process_file(my_file)
# File remains open (nullcontext doesn't close it)
closing
The closing context manager ensures an object gets its close() method called when the block exits. This is useful for objects that don’t support the context manager protocol but have a close() method.
Signature
contextlib.closing(thing)
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
thing | object | — | Object with a close() method |
Examples
from contextlib import closing
from urllib.request import urlopen
# urlopen doesn't support context manager in older Python
with closing(urlopen('https://example.com')) as page:
content = page.read()
# page.close() is called automatically here
# Another example: objects with close but no context manager
class DatabaseConnection:
def __init__(self, name):
self.name = name
self.connected = True
def close(self):
self.connected = False
print(f"Closed {self.name}")
with closing(DatabaseConnection("users")) as db:
print(f"Using {db.name}")
# Output: Using users
# Output: Closed users
redirect_stdout and redirect_stderr
These context managers temporarily redirect stdout or stderr to a different file-like object. They’re useful for capturing output in tests or sending output elsewhere.
Signatures
contextlib.redirect_stdout(new_target)
contextlib.redirect_stderr(new_target)
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
new_target | file-like | — | File or StringIO to redirect to |
Examples
import sys
from io import StringIO
from contextlib import redirect_stdout, redirect_stderr
# Capture stdout to a string
output = StringIO()
with redirect_stdout(output):
print("Hello from stdout")
print("Another line")
result = output.getvalue()
print(f"Captured: {repr(result)}")
# Output: Captured: 'Hello from stdout\nAnother line\n'
# Capture stderr
error_output = StringIO()
with redirect_stderr(error_output):
sys.stderr.write("Error message here\n")
errors = error_output.getvalue()
print(f"Errors: {repr(errors)}")
# Output: Errors: 'Error message here\n'
ExitStack
The ExitStack class manages multiple context managers in a single block. Instead of nesting with statements, you register cleanup callbacks and they all run when the block exits—even if exceptions occur.
Signature
contextlib.ExitStack()
Key Methods
| Method | Description |
|---|---|
enter_context(cm) | Enter a context manager and register its __exit__ |
push(exit) | Register a callback or context manager’s __exit__ |
callback(fn, *args) | Register a function to call on exit |
pop_all() | Transfer callbacks to a new ExitStack |
close() | Explicitly run all callbacks |
Examples
from contextlib import ExitStack
# Opening multiple files safely
filenames = ['a.txt', 'b.txt', 'c.txt']
with ExitStack() as stack:
files = [stack.enter_context(open(name, 'w')) for name in filenames]
for f, name in zip(files, filenames):
f.write(f"Content from {name}")
# All files closed automatically
# Dynamic cleanup with callbacks
with ExitStack() as stack:
# Acquire resources
conn = acquire_connection()
lock = acquire_lock()
# Register cleanups (runs in reverse order)
stack.callback(release_lock, lock)
stack.callback(close_connection, conn)
do_work(conn)
# Cleanup runs automatically
Conditional cleanup with pop_all
from contextlib import ExitStack
def get_file_contents(filenames, keep_open=False):
with ExitStack() as stack:
files = [stack.enter_context(open(f)) for f in filenames]
contents = [f.read() for f in files]
if keep_open:
# Transfer callbacks to new stack - caller responsible
return contents, stack.pop_all().close
# Otherwise files close normally
return contents
# Keep files open after function returns
data, closer = get_file_contents(['data.txt'], keep_open=True)
# Files are still open here
closer() # Caller explicitly closes them
AsyncExitStack
The async version of ExitStack handles both sync and async context managers. It also supports coroutines for cleanup logic.
Key Methods
| Method | Description |
|---|---|
enter_async_context(cm) | Enter an async context manager |
push_async_exit(exit) | Register async exit callback |
push_async_callback(fn) | Register async function |
aclose() | Run all async callbacks |
Examples
import asyncio
from contextlib import AsyncExitStack, asynccontextmanager
@asynccontextmanager
async def get_connection(name):
print(f"Connecting to {name}")
yield f"conn_{name}"
print(f"Disconnected {name}")
async def main():
async with AsyncExitStack() as stack:
# Multiple async connections
conn1 = await stack.enter_async_context(get_connection("db"))
conn2 = await stack.enter_async_context(get_connection("cache"))
print(f"Using {conn1} and {conn2}")
# Both connections closed automatically
asyncio.run(main())
# Output:
# Connecting to db
# Connecting to cache
# Using conn_db and conn_cache
# Disconnected cache
# Disconnected db
AbstractContextManager
A base class for creating context managers via inheritance. It provides a default __enter__ that returns self, while __exit__ remains abstract.
Examples
from contextlib import AbstractContextManager
class Timer(AbstractContextManager):
def __init__(self):
self.elapsed = 0
def __enter__(self):
import time
self.start = time.monotonic()
return self
def __exit__(self, *exc):
import time
self.elapsed = time.monotonic() - self.start
return False
with Timer() as t:
sum(range(1000000))
print(f"Elapsed: {t.elapsed:.4f}s")
# Output: Elapsed: 0.0034s (varies)
ContextDecorator
A base class that enables using context managers as function decorators. When you inherit from it, your context manager works both in with statements and as a decorator.
Examples
from contextlib import ContextDecorator
import time
class timing(ContextDecorator):
def __init__(self, name):
self.name = name
def __enter__(self):
self.start = time.time()
return self
def __exit__(self, *args):
elapsed = time.time() - self.start
print(f"{self.name} took {elapsed:.3f}s")
# As a context manager
with timing("block"):
sum(range(1000000))
# As a decorator
@timing("function")
def slow_function():
time.sleep(0.1)
slow_function()
# Output:
# block took 0.003s
# function took 0.101s
Common Patterns
Optional context manager
from contextlib import nullcontext, suppress
def process(requires_lock=False):
# Conditionally use a context manager
cm = suppress(FileNotFoundError) if requires_lock else nullcontext()
with cm:
# Do something that might fail
pass
Retry with cleanup
from contextlib import ExitStack, contextmanager
@contextmanager
def retry_on_failure(max_attempts=3):
attempts = 0
with ExitStack() as stack:
while attempts < max_attempts:
try:
resource = acquire_resource()
stack.callback(release_resource, resource)
yield resource
return # Success - keep resource
except Exception as e:
attempts += 1
if attempts == max_attempts:
raise
print(f"Attempt {attempts} failed, retrying...")
stack.pop_all() # Don't clean up on failure - caller handles
Redirecting during tests
from io import StringIO
from contextlib import redirect_stdout
import sys
def test_print_function():
output = StringIO()
with redirect_stdout(output):
print("test output")
assert output.getvalue() == "test output\n"