Advanced Context Techniques
Nested Context Managers
Nested context managers allow you to manage multiple resources simultaneously:
from contextlib import ExitStack
def process_multiple_resources():
with ExitStack() as stack:
## Dynamically enter multiple context managers
files = [stack.enter_context(open(f'/tmp/file{i}.txt', 'w'))
for i in range(3)]
## Perform operations on multiple files
for file in files:
file.write(f'Content for {file.name}')
Context Manager State Machine
graph TD
A[Initialization] --> B[Enter Context]
B --> C{Resource State}
C -->|Success| D[Execute Code]
C -->|Error| E[Handle Exception]
D --> F[Exit Context]
E --> F
F --> G[Cleanup Resources]
Parameterized Context Managers
class RetryOperation:
def __init__(self, max_attempts=3):
self.max_attempts = max_attempts
self.current_attempt = 0
def __enter__(self):
self.current_attempt += 1
return self
def __exit__(self, exc_type, exc_value, traceback):
if exc_type is not None and self.current_attempt < self.max_attempts:
print(f'Retry attempt {self.current_attempt}')
return True
return False
## Usage
with RetryOperation(max_attempts=5):
## Potentially failing operation
result = risky_network_call()
Contextlib Advanced Techniques
Suppressing Exceptions
from contextlib import suppress
## Ignore specific exceptions
with suppress(FileNotFoundError):
os.remove('/tmp/nonexistent_file.txt')
Reusable Context Managers
from contextlib import contextmanager
@contextmanager
def managed_temporary_directory():
import tempfile
import shutil
temp_dir = tempfile.mkdtemp()
try:
yield temp_dir
finally:
shutil.rmtree(temp_dir)
## Usage
with managed_temporary_directory() as tmpdir:
## Perform operations in temporary directory
with open(f'{tmpdir}/test.txt', 'w') as f:
f.write('Temporary file content')
Context Manager Comparison
Technique |
Use Case |
Complexity |
Basic with |
Simple resource management |
Low |
ExitStack |
Dynamic resource handling |
Medium |
Parameterized CMs |
Configurable behavior |
Medium |
contextlib Decorators |
Simplified implementation |
Low |
Asynchronous Context Managers
import asyncio
class AsyncResourceManager:
async def __aenter__(self):
print('Entering async context')
return self
async def __aexit__(self, exc_type, exc_value, traceback):
print('Exiting async context')
async def main():
async with AsyncResourceManager() as manager:
## Asynchronous operations
await asyncio.sleep(1)
## Run async context manager
asyncio.run(main())
- Minimize overhead in context manager methods
- Avoid complex logic in
__enter__
and __exit__
- Use built-in context managers when possible
- Profile and optimize resource-intensive contexts
Error Handling Strategies
class RobustContextManager:
def __enter__(self):
## Validate initial state
if not self.is_ready():
raise RuntimeError('Resource not ready')
return self
def __exit__(self, exc_type, exc_value, traceback):
## Log exceptions
if exc_type:
print(f'Exception occurred: {exc_type}')
## Always perform cleanup
self.cleanup()
## Optionally suppress specific exceptions
return exc_type is ValueError
By mastering these advanced context management techniques, you can create more robust, flexible, and efficient Python applications with LabEx-level resource handling strategies.