mirror of
https://github.com/blshaer/python-by-example.git
synced 2026-03-27 23:29:25 +01:00
383 lines
11 KiB
Python
383 lines
11 KiB
Python
"""
|
|
================================================================================
|
|
File: 04_context_managers.py
|
|
Topic: Context Managers in Python
|
|
================================================================================
|
|
|
|
This file demonstrates context managers in Python. Context managers are objects
|
|
that define setup and cleanup actions, typically used with the 'with' statement.
|
|
They ensure resources are properly managed even when errors occur.
|
|
|
|
Key Concepts:
|
|
- The 'with' statement
|
|
- __enter__ and __exit__ methods
|
|
- contextlib module
|
|
- File handling
|
|
- Resource management patterns
|
|
|
|
================================================================================
|
|
"""
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# 1. The 'with' Statement
|
|
# -----------------------------------------------------------------------------
|
|
# Ensures cleanup happens automatically
|
|
|
|
print("--- The 'with' Statement ---")
|
|
|
|
# Most common use: file handling
|
|
# Without 'with' (error-prone)
|
|
"""
|
|
file = open('example.txt', 'w')
|
|
try:
|
|
file.write('Hello')
|
|
finally:
|
|
file.close() # Must remember to close!
|
|
"""
|
|
|
|
# With 'with' (automatic cleanup)
|
|
"""
|
|
with open('example.txt', 'w') as file:
|
|
file.write('Hello')
|
|
# File is automatically closed here
|
|
"""
|
|
|
|
print("The 'with' statement:")
|
|
print(" - Automatically handles setup and cleanup")
|
|
print(" - Ensures cleanup even if errors occur")
|
|
print(" - Cleaner, more readable code")
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# 2. Creating a Context Manager Class
|
|
# -----------------------------------------------------------------------------
|
|
# Implement __enter__ and __exit__
|
|
|
|
print("\n--- Context Manager Class ---")
|
|
|
|
class ManagedFile:
|
|
"""A custom context manager for file handling."""
|
|
|
|
def __init__(self, filename, mode='r'):
|
|
self.filename = filename
|
|
self.mode = mode
|
|
self.file = None
|
|
|
|
def __enter__(self):
|
|
"""Called when entering 'with' block."""
|
|
print(f" Opening file: {self.filename}")
|
|
# In real code: self.file = open(self.filename, self.mode)
|
|
self.file = f"<FileObject: {self.filename}>"
|
|
return self.file # This is assigned to the 'as' variable
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
"""Called when exiting 'with' block."""
|
|
print(f" Closing file: {self.filename}")
|
|
# In real code: self.file.close()
|
|
|
|
# exc_type, exc_val, exc_tb contain exception info if an error occurred
|
|
if exc_type is not None:
|
|
print(f" Exception occurred: {exc_type.__name__}: {exc_val}")
|
|
|
|
# Return True to suppress the exception, False to propagate
|
|
return False
|
|
|
|
# Using the context manager
|
|
print("Normal usage:")
|
|
with ManagedFile("data.txt", "w") as f:
|
|
print(f" Working with: {f}")
|
|
|
|
print("\nWith exception:")
|
|
try:
|
|
with ManagedFile("data.txt") as f:
|
|
print(f" Working with: {f}")
|
|
raise ValueError("Something went wrong!")
|
|
except ValueError:
|
|
print(" Exception was propagated")
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# 3. Context Manager with Exception Suppression
|
|
# -----------------------------------------------------------------------------
|
|
|
|
print("\n--- Suppressing Exceptions ---")
|
|
|
|
class SuppressErrors:
|
|
"""Context manager that suppresses specified exceptions."""
|
|
|
|
def __init__(self, *exceptions):
|
|
self.exceptions = exceptions
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
if exc_type is not None and issubclass(exc_type, self.exceptions):
|
|
print(f" Suppressed: {exc_type.__name__}: {exc_val}")
|
|
return True # Suppress the exception
|
|
return False # Don't suppress
|
|
|
|
# Using the suppressor
|
|
print("Suppressing ValueError:")
|
|
with SuppressErrors(ValueError, TypeError):
|
|
print(" Before error")
|
|
raise ValueError("This will be suppressed")
|
|
print(" After error") # Never reached
|
|
|
|
print(" Code continues after 'with' block")
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# 4. Using contextlib
|
|
# -----------------------------------------------------------------------------
|
|
# Create context managers without classes
|
|
|
|
print("\n--- Using contextlib ---")
|
|
|
|
from contextlib import contextmanager
|
|
|
|
@contextmanager
|
|
def managed_resource(name):
|
|
"""A context manager using @contextmanager decorator."""
|
|
print(f" Acquiring: {name}")
|
|
try:
|
|
yield name # Control is passed to 'with' block here
|
|
finally:
|
|
print(f" Releasing: {name}")
|
|
|
|
print("@contextmanager decorator:")
|
|
with managed_resource("Database") as resource:
|
|
print(f" Using: {resource}")
|
|
|
|
# Timer context manager
|
|
import time
|
|
|
|
@contextmanager
|
|
def timer(description="Operation"):
|
|
"""Time the enclosed code block."""
|
|
start = time.time()
|
|
try:
|
|
yield
|
|
finally:
|
|
elapsed = time.time() - start
|
|
print(f" {description} took {elapsed:.4f} seconds")
|
|
|
|
print("\nTimer context manager:")
|
|
with timer("Sleep operation"):
|
|
time.sleep(0.1)
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# 5. Multiple Context Managers
|
|
# -----------------------------------------------------------------------------
|
|
|
|
print("\n--- Multiple Context Managers ---")
|
|
|
|
@contextmanager
|
|
def open_mock_file(name):
|
|
"""Mock file context manager."""
|
|
print(f" Opening: {name}")
|
|
yield f"<{name}>"
|
|
print(f" Closing: {name}")
|
|
|
|
# Multiple context managers in one 'with'
|
|
print("Multiple context managers:")
|
|
with open_mock_file("input.txt") as infile, \
|
|
open_mock_file("output.txt") as outfile:
|
|
print(f" Reading from: {infile}")
|
|
print(f" Writing to: {outfile}")
|
|
|
|
# Python 3.10+ allows parenthesized context managers
|
|
"""
|
|
with (
|
|
open_mock_file("input.txt") as infile,
|
|
open_mock_file("output.txt") as outfile
|
|
):
|
|
...
|
|
"""
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# 6. Practical Context Managers
|
|
# -----------------------------------------------------------------------------
|
|
|
|
print("\n--- Practical Context Managers ---")
|
|
|
|
# 1. Database connection (mock)
|
|
@contextmanager
|
|
def database_connection(host):
|
|
"""Mock database connection."""
|
|
print(f" Connecting to {host}...")
|
|
connection = {"host": host, "connected": True}
|
|
try:
|
|
yield connection
|
|
finally:
|
|
connection["connected"] = False
|
|
print(f" Disconnected from {host}")
|
|
|
|
print("Database connection:")
|
|
with database_connection("localhost") as db:
|
|
print(f" Connected: {db['connected']}")
|
|
# Perform database operations
|
|
|
|
# 2. Temporary working directory
|
|
import os
|
|
from contextlib import contextmanager
|
|
|
|
@contextmanager
|
|
def temp_directory_context():
|
|
"""Change to temp directory and restore."""
|
|
original_dir = os.getcwd()
|
|
temp_dir = os.path.dirname(original_dir) if original_dir else original_dir
|
|
try:
|
|
# In real code: os.chdir(temp_dir)
|
|
print(f" Changed to: {temp_dir}")
|
|
yield temp_dir
|
|
finally:
|
|
# os.chdir(original_dir)
|
|
print(f" Restored to: {original_dir}")
|
|
|
|
print("\nDirectory change:")
|
|
with temp_directory_context():
|
|
print(" Working in temp directory")
|
|
|
|
# 3. Lock context manager
|
|
@contextmanager
|
|
def locked(lock_name="default"):
|
|
"""Mock lock context manager."""
|
|
print(f" Acquiring lock: {lock_name}")
|
|
try:
|
|
yield
|
|
finally:
|
|
print(f" Releasing lock: {lock_name}")
|
|
|
|
print("\nLocking:")
|
|
with locked("resource_lock"):
|
|
print(" Critical section")
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# 7. contextlib Utilities
|
|
# -----------------------------------------------------------------------------
|
|
|
|
print("\n--- contextlib Utilities ---")
|
|
|
|
from contextlib import closing, suppress
|
|
|
|
# closing() - for objects with close() but no __exit__
|
|
class Connection:
|
|
def close(self):
|
|
print(" Connection closed")
|
|
|
|
print("using closing():")
|
|
with closing(Connection()) as conn:
|
|
print(" Using connection")
|
|
|
|
# suppress() - ignore specified exceptions
|
|
print("\nsuppress() example:")
|
|
from contextlib import suppress
|
|
|
|
# Instead of try/except for simple cases
|
|
with suppress(FileNotFoundError, KeyError):
|
|
# This would normally raise an error
|
|
print(" Attempting risky operations...")
|
|
# raise FileNotFoundError("No such file")
|
|
print(" Continued after suppress")
|
|
|
|
# nullcontext() - do-nothing context manager (Python 3.7+)
|
|
from contextlib import nullcontext
|
|
|
|
def process_data(data, lock=None):
|
|
"""Process with optional lock."""
|
|
with lock if lock else nullcontext():
|
|
print(f" Processing: {data}")
|
|
|
|
print("\nnullcontext() example:")
|
|
process_data("data1") # No lock
|
|
process_data("data2", locked("my_lock")) # With lock
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# 8. Async Context Managers
|
|
# -----------------------------------------------------------------------------
|
|
|
|
print("\n--- Async Context Managers ---")
|
|
|
|
# For async code, use __aenter__ and __aexit__
|
|
async_example = '''
|
|
class AsyncResource:
|
|
async def __aenter__(self):
|
|
print("Acquiring async resource...")
|
|
await asyncio.sleep(0.1)
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
|
print("Releasing async resource...")
|
|
await asyncio.sleep(0.1)
|
|
return False
|
|
|
|
# Usage:
|
|
async with AsyncResource() as resource:
|
|
print("Using resource")
|
|
'''
|
|
|
|
print("Async context managers use __aenter__ and __aexit__")
|
|
print("Use 'async with' statement")
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# 9. Reentrant and Reusable Context Managers
|
|
# -----------------------------------------------------------------------------
|
|
|
|
print("\n--- Reentrant Context Managers ---")
|
|
|
|
# A context manager that can be used multiple times
|
|
class ReusableContext:
|
|
"""Context manager that can be reused."""
|
|
|
|
def __init__(self, name):
|
|
self.name = name
|
|
self.uses = 0
|
|
|
|
def __enter__(self):
|
|
self.uses += 1
|
|
print(f" Entering {self.name} (use #{self.uses})")
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
print(f" Exiting {self.name}")
|
|
return False
|
|
|
|
ctx = ReusableContext("MyContext")
|
|
|
|
print("Reusing context manager:")
|
|
with ctx:
|
|
print(" First use")
|
|
|
|
with ctx:
|
|
print(" Second use")
|
|
|
|
print(f"Total uses: {ctx.uses}")
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# 10. Real-World Pattern: Configuration Override
|
|
# -----------------------------------------------------------------------------
|
|
|
|
print("\n--- Real-World: Config Override ---")
|
|
|
|
class Config:
|
|
"""Application configuration."""
|
|
settings = {"debug": False, "log_level": "INFO"}
|
|
|
|
@classmethod
|
|
@contextmanager
|
|
def override(cls, **overrides):
|
|
"""Temporarily override configuration."""
|
|
original = cls.settings.copy()
|
|
try:
|
|
cls.settings.update(overrides)
|
|
print(f" Config overridden: {overrides}")
|
|
yield cls.settings
|
|
finally:
|
|
cls.settings = original
|
|
print(" Config restored")
|
|
|
|
print(f"Original config: {Config.settings}")
|
|
|
|
with Config.override(debug=True, log_level="DEBUG") as settings:
|
|
print(f" Inside 'with': {settings}")
|
|
|
|
print(f"After 'with': {Config.settings}")
|