Python's contextlib module provides utilities for working with context managers, which are used for resource management (e.g., opening/closing files, acquiring/releasing locks). You can create custom context managers using contextlib, making your code cleaner and more efficient. Below are examples of how to use contextlib to create and manage custom context managers.
1. Using contextlib.contextmanager to Create a Context Manager
The @contextmanager decorator allows you to create a context manager with a generator.
Copy
from contextlib import contextmanager
@contextmanager
def open_file(file_name, mode):
file = open(file_name, mode)
try:
yield file
finally:
file.close()
print("File closed.")
# Usage
with open_file("example.txt", "w") as f:
f.write("Hello, contextlib!")
Explanation:
The yield statement pauses the function and provides control to the with block.
Code after yield ensures cleanup (e.g., closing the file).
2. Custom Context Manager for Logging
Use contextlib.contextmanager to implement a logging context manager.
Copy
Output:
Copy
3. Suppressing Exceptions with contextlib.suppress
The suppress context manager can ignore specified exceptions.
Copy
Explanation:
The suppress context manager ignores the specified exception (FileNotFoundError here).
4. Temporary Change with contextlib
Use @contextmanager to temporarily change a variable or system state.
Copy
5. Redirecting Standard Output with redirect_stdout
Redirect the standard output to a file or other writable objects.
Copy
6. Closing Resources Automatically
contextlib.closing is useful when working with objects that have a .close() method.
Copy
7. Nested Context Managers with contextlib.ExitStack
ExitStack allows you to manage multiple context managers dynamically.
Copy
Explanation:
ExitStack ensures all opened files are closed, even if an exception occurs.
8. Using contextlib.AsyncExitStack for Async Context Managers
Manage multiple asynchronous context managers.
Copy
9. Custom Context Manager Using contextlib.AbstractContextManager
Define a custom context manager class.
Copy
10. Timeout Context Manager
Implement a context manager for timing out operations.
Copy
Key Features of contextlib:
Efficient Resource Management: Handles setup and teardown cleanly.
from contextlib import contextmanager
@contextmanager
def log_context(message):
print(f"Start: {message}")
try:
yield
finally:
print(f"End: {message}")
# Usage
with log_context("Logging process"):
print("Running the process...")
Start: Logging process
Running the process...
End: Logging process
from contextlib import suppress
# Usage
with suppress(FileNotFoundError):
with open("non_existent_file.txt") as f:
print(f.read())
print("No error raised!")
from contextlib import contextmanager
@contextmanager
def change_value(obj, attr, new_value):
old_value = getattr(obj, attr)
setattr(obj, attr, new_value)
try:
yield
finally:
setattr(obj, attr, old_value)
# Usage
class Example:
value = 10
example = Example()
print(f"Before: {example.value}")
with change_value(example, "value", 20):
print(f"Inside context: {example.value}")
print(f"After: {example.value}")
from contextlib import redirect_stdout
import io
buffer = io.StringIO()
# Redirect stdout
with redirect_stdout(buffer):
print("This will be written to the buffer.")
print("Captured:", buffer.getvalue())
from contextlib import closing
import urllib.request
# Automatically close the URL resource
with closing(urllib.request.urlopen("https://example.com")) as page:
content = page.read()
print("Content length:", len(content))
from contextlib import ExitStack
# Manage multiple files dynamically
with ExitStack() as stack:
files = [stack.enter_context(open(f"file_{i}.txt", "w")) for i in range(3)]
for i, f in enumerate(files):
f.write(f"File {i} content\n")
import asyncio
from contextlib import AsyncExitStack
class AsyncResource:
async def __aenter__(self):
print("Acquiring resource")
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
print("Releasing resource")
async def main():
async with AsyncExitStack() as stack:
resource = await stack.enter_async_context(AsyncResource())
print("Using resource")
asyncio.run(main())
from contextlib import AbstractContextManager
class CustomContext(AbstractContextManager):
def __enter__(self):
print("Entering context")
return self
def __exit__(self, exc_type, exc_value, traceback):
print("Exiting context")
# Usage
with CustomContext():
print("Inside the context")
import signal
from contextlib import contextmanager
@contextmanager
def timeout(seconds):
def handler(signum, frame):
raise TimeoutError("Operation timed out!")
signal.signal(signal.SIGALRM, handler)
signal.alarm(seconds)
try:
yield
finally:
signal.alarm(0)
# Usage
try:
with timeout(2):
while True: # Simulate long-running operation
pass
except TimeoutError as e:
print(e)