Bài 4: Context Managers
Mục Tiêu Bài Học
Sau khi hoàn thành bài này, bạn sẽ:
- ✅ Hiểu context managers và with statement
- ✅ Tạo context managers với
__enter__và__exit__ - ✅ Sử dụng contextlib module
- ✅ Áp dụng context managers cho resource management
- ✅ Tạo reusable context managers
- ✅ Xử lý exceptions trong context managers
Context Managers Là Gì?
Context Manager là object quản lý resources và đảm bảo cleanup code luôn được thực thi.
The Problem
# Without context manager - risky!file = open('data.txt', 'r')content = file.read()# What if exception happens here?file.close() # Might never execute! # Better - but verbosefile = open('data.txt', 'r')try: content = file.read()finally: file.close() # Always executes
The Solution - with Statement
# With context manager - clean and safe!with open('data.txt', 'r') as file: content = file.read()# file.close() automatically called! # Multiple context managerswith open('input.txt', 'r') as infile, open('output.txt', 'w') as outfile: content = infile.read() outfile.write(content.upper())# Both files closed automatically
Context Manager Protocol
Context manager implement 2 methods: __enter__ và __exit__.
Basic Structure
class MyContextManager: """Basic context manager.""" def __enter__(self): """ Called when entering 'with' block. Return value is assigned to 'as' variable. """ print("Entering context") return self # Or any object def __exit__(self, exc_type, exc_value, traceback): """ Called when exiting 'with' block. Args: exc_type: Exception class (or None) exc_value: Exception instance (or None) traceback: Traceback object (or None) Returns: True: Suppress exception False/None: Propagate exception """ print("Exiting context") return False # Don't suppress exceptions # Usagewith MyContextManager() as cm: print("Inside context")# Output:# Entering context# Inside context# Exiting context
How it Works
# This code:with context_manager as var: # code block pass # Is equivalent to:var = context_manager.__enter__()try: # code block passfinally: context_manager.__exit__(None, None, None)
Custom Context Managers
1. File Handler
class FileHandler: """Custom file context manager.""" def __init__(self, filename, mode='r'): self.filename = filename self.mode = mode self.file = None def __enter__(self): print(f"Opening {self.filename}") self.file = open(self.filename, self.mode) return self.file def __exit__(self, exc_type, exc_value, traceback): print(f"Closing {self.filename}") if self.file: self.file.close() return False # Usagewith FileHandler('test.txt', 'w') as f: f.write("Hello, World!")# Opening test.txt# Closing test.txt # Verify file was writtenwith open('test.txt', 'r') as f: print(f.read()) # Hello, World!
2. Timer Context Manager
import time class Timer: """Measure execution time.""" def __init__(self, name="Operation"): self.name = name self.start_time = None self.elapsed = None def __enter__(self): print(f"Starting {self.name}...") self.start_time = time.time() return self def __exit__(self, exc_type, exc_value, traceback): self.elapsed = time.time() - self.start_time print(f"{self.name} took {self.elapsed:.4f} seconds") return False # Usagewith Timer("Data processing"): # Simulate work time.sleep(1) total = sum(range(1000000))# Starting Data processing...# Data processing took 1.0234 seconds # Access elapsed timewith Timer("Calculation") as timer: result = sum(range(1000000)) print(f"Elapsed: {timer.elapsed:.4f}s")
3. Database Connection
class DatabaseConnection: """Database connection context manager.""" def __init__(self, host, database): self.host = host self.database = database self.connection = None def __enter__(self): print(f"Connecting to {self.database} on {self.host}") # Simulate connection self.connection = { 'host': self.host, 'database': self.database, 'connected': True } return self.connection def __exit__(self, exc_type, exc_value, traceback): print(f"Closing connection to {self.database}") if self.connection: self.connection['connected'] = False return False # Usagewith DatabaseConnection('localhost', 'mydb') as conn: print(f"Connected: {conn['connected']}") # Execute queries...# Connecting to mydb on localhost# Connected: True# Closing connection to mydb
4. Directory Changer
import os class ChangeDirectory: """Temporarily change working directory.""" def __init__(self, path): self.path = path self.old_path = None def __enter__(self): self.old_path = os.getcwd() print(f"Changing directory: {self.old_path} -> {self.path}") os.chdir(self.path) return self.path def __exit__(self, exc_type, exc_value, traceback): print(f"Restoring directory: {self.path} -> {self.old_path}") os.chdir(self.old_path) return False # Usageprint(f"Current: {os.getcwd()}") with ChangeDirectory('/tmp'): print(f"Inside with: {os.getcwd()}") # Do work in /tmp print(f"After with: {os.getcwd()}")
5. Suppressing Exceptions
class SuppressException: """Suppress specific exceptions.""" def __init__(self, *exceptions): self.exceptions = exceptions def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): # Return True to suppress exception if exc_type is not None: if issubclass(exc_type, self.exceptions): print(f"Suppressed {exc_type.__name__}: {exc_value}") return True # Suppress return False # Don't suppress # Usageprint("Before") with SuppressException(ValueError, TypeError): print("Inside") raise ValueError("This error is suppressed") print("This won't execute") print("After - execution continues!")# Before# Inside# Suppressed ValueError: This error is suppressed# After - execution continues! # Unsuppressed exceptiontry: with SuppressException(ValueError): raise KeyError("This will propagate")except KeyError as e: print(f"Caught: {e}")
contextlib Module
Module contextlib cung cấp utilities để tạo context managers dễ hơn.
@contextmanager Decorator
from contextlib import contextmanager @contextmanagerdef file_handler(filename, mode='r'): """Simple file context manager.""" print(f"Opening {filename}") file = open(filename, mode) try: yield file # Value returned to 'as' variable finally: print(f"Closing {filename}") file.close() # Usagewith file_handler('test.txt', 'w') as f: f.write("Using contextmanager decorator!")# Opening test.txt# Closing test.txt
Timer với @contextmanager
import timefrom contextlib import contextmanager @contextmanagerdef timer(name="Operation"): """Timer context manager.""" print(f"Starting {name}...") start = time.time() try: yield finally: elapsed = time.time() - start print(f"{name} took {elapsed:.4f} seconds") # Usagewith timer("Processing"): time.sleep(0.5) result = sum(range(1000000))# Starting Processing...# Processing took 0.5234 seconds
Temporary State Change
from contextlib import contextmanager class Config: """Simple configuration.""" debug = False @contextmanagerdef debug_mode(): """Temporarily enable debug mode.""" old_value = Config.debug Config.debug = True print("Debug mode: ON") try: yield finally: Config.debug = old_value print("Debug mode: OFF") # Usageprint(f"Debug: {Config.debug}") # False with debug_mode(): print(f"Debug: {Config.debug}") # True # Debug operations... print(f"Debug: {Config.debug}") # False (restored)
suppress()
from contextlib import suppress # Suppress specific exceptionswith suppress(FileNotFoundError): with open('nonexistent.txt', 'r') as f: content = f.read()# No exception raised! print("Execution continues") # Multiple exceptionswith suppress(ValueError, TypeError, KeyError): # Code that might raise these exceptions result = int("not a number")# Silently suppressed
redirect_stdout() và redirect_stderr()
from contextlib import redirect_stdout, redirect_stderrimport io # Redirect stdoutoutput = io.StringIO() with redirect_stdout(output): print("This goes to StringIO") print("Not to console!") print("Back to console")print(f"Captured: {output.getvalue()}")# Back to console# Captured: This goes to StringIO# Not to console! # Redirect to filewith open('output.txt', 'w') as f: with redirect_stdout(f): print("Written to file") print("Not to console") with open('output.txt', 'r') as f: print(f"File content: {f.read()}")
ExitStack
from contextlib import ExitStack def process_files(filenames): """Process multiple files with ExitStack.""" with ExitStack() as stack: # Open all files files = [stack.enter_context(open(fn, 'r')) for fn in filenames] # Process files for i, file in enumerate(files): print(f"File {i}: {file.readline().strip()}") # All files closed automatically # Create test filesfor i in range(3): with open(f'file{i}.txt', 'w') as f: f.write(f"Content of file {i}\n") # Processprocess_files(['file0.txt', 'file1.txt', 'file2.txt'])# File 0: Content of file 0# File 1: Content of file 1# File 2: Content of file 2
Conditional Context Manager
from contextlib import contextmanager, nullcontext @contextmanagerdef optional_context(condition, value): """Context manager that's conditional.""" if condition: print(f"Using context with {value}") yield value else: yield None # Usagefor use_context in [True, False]: ctx = optional_context(use_context, "data") if use_context else nullcontext() with ctx: print("Inside context") print()
Real-world Examples
1. Database Transaction
from contextlib import contextmanager class Database: """Database with transactions.""" def __init__(self): self.in_transaction = False self.data = {} def begin(self): self.in_transaction = True self.backup = self.data.copy() print("Transaction started") def commit(self): self.in_transaction = False self.backup = None print("Transaction committed") def rollback(self): self.data = self.backup.copy() self.in_transaction = False self.backup = None print("Transaction rolled back") @contextmanager def transaction(self): """Transaction context manager.""" self.begin() try: yield self self.commit() except Exception as e: self.rollback() raise # Usagedb = Database() # Successful transactionwith db.transaction(): db.data['user1'] = 'Alice' db.data['user2'] = 'Bob'# Transaction started# Transaction committed print(db.data) # {'user1': 'Alice', 'user2': 'Bob'} # Failed transactiontry: with db.transaction(): db.data['user3'] = 'Charlie' raise ValueError("Oops!")except ValueError: pass# Transaction started# Transaction rolled back print(db.data) # {'user1': 'Alice', 'user2': 'Bob'} (unchanged)
2. Lock Manager
import threadingfrom contextlib import contextmanager class Resource: """Shared resource with locking.""" def __init__(self): self.lock = threading.Lock() self.value = 0 @contextmanager def locked(self): """Lock context manager.""" print(f"Thread {threading.current_thread().name}: Acquiring lock") self.lock.acquire() try: yield self finally: print(f"Thread {threading.current_thread().name}: Releasing lock") self.lock.release() # Usageresource = Resource() def worker(resource, iterations): """Worker thread.""" for _ in range(iterations): with resource.locked(): resource.value += 1 # Single-threaded demowith resource.locked(): resource.value += 10 print(f"Value: {resource.value}")# Thread MainThread: Acquiring lock# Value: 10# Thread MainThread: Releasing lock
3. Environment Variables
import osfrom contextlib import contextmanager @contextmanagerdef env_variable(key, value): """Temporarily set environment variable.""" old_value = os.environ.get(key) if value is None: os.environ.pop(key, None) else: os.environ[key] = value print(f"Set {key}={value}") try: yield finally: if old_value is None: os.environ.pop(key, None) else: os.environ[key] = old_value print(f"Restored {key}={old_value}") # Usageprint(f"Before: {os.environ.get('TEST_VAR', 'Not set')}") with env_variable('TEST_VAR', 'temporary_value'): print(f"Inside: {os.environ.get('TEST_VAR')}") print(f"After: {os.environ.get('TEST_VAR', 'Not set')}")# Before: Not set# Set TEST_VAR=temporary_value# Inside: temporary_value# Restored TEST_VAR=None# After: Not set
4. Mock/Patch Context
from contextlib import contextmanager @contextmanagerdef mock_function(module, func_name, mock_impl): """Mock a function temporarily.""" original = getattr(module, func_name) setattr(module, func_name, mock_impl) print(f"Mocked {module.__name__}.{func_name}") try: yield finally: setattr(module, func_name, original) print(f"Restored {module.__name__}.{func_name}") # Example moduleclass MyModule: @staticmethod def get_data(): return "Real data" # Usageprint(MyModule.get_data()) # Real data with mock_function(MyModule, 'get_data', lambda: "Mock data"): print(MyModule.get_data()) # Mock data print(MyModule.get_data()) # Real data (restored)
5. Profiler Context
import timefrom contextlib import contextmanager @contextmanagerdef profiler(name="Code block"): """Profile code execution.""" stats = { 'name': name, 'start_time': None, 'end_time': None, 'elapsed': None } print(f"Profiling: {name}") stats['start_time'] = time.time() try: yield stats finally: stats['end_time'] = time.time() stats['elapsed'] = stats['end_time'] - stats['start_time'] print(f"\nProfile Results for '{name}':") print(f" Start: {stats['start_time']:.2f}") print(f" End: {stats['end_time']:.2f}") print(f" Elapsed: {stats['elapsed']:.4f}s") # Usagewith profiler("Data processing") as prof: time.sleep(0.5) result = sum(range(1000000)) time.sleep(0.3)# Profiling: Data processing## Profile Results for 'Data processing':# Start: 1698400000.12# End: 1698400000.92# Elapsed: 0.8001s
Nested Context Managers
from contextlib import contextmanager @contextmanagerdef operation(name): """Operation context.""" print(f"Start {name}") try: yield finally: print(f"End {name}") # Nested manuallywith operation("Outer"): with operation("Middle"): with operation("Inner"): print("Core operation")# Start Outer# Start Middle# Start Inner# Core operation# End Inner# End Middle# End Outer # Multiple on one linewith operation("A"), operation("B"), operation("C"): print("Core operation")
Best Practices
# 1. Always clean up in __exit__ or finally@contextmanagerdef resource_manager(): resource = acquire_resource() try: yield resource finally: release_resource(resource) # Always executes # 2. Return False from __exit__ unless suppressingdef __exit__(self, exc_type, exc_value, traceback): self.cleanup() return False # Let exceptions propagate # 3. Use contextlib for simple casesfrom contextlib import contextmanager @contextmanagerdef simple_context(): # Setup try: yield finally: # Cleanup pass # 4. Document exception handling@contextmanagerdef documented_context(): """ Context manager that handles resources. Exceptions are propagated after cleanup. """ pass # 5. Use ExitStack for multiple resourcesfrom contextlib import ExitStack with ExitStack() as stack: files = [stack.enter_context(open(f)) for f in filenames] # All closed automatically
Bài Tập Thực Hành
Bài 1: Logging Context
Tạo context manager log entry/exit với indentation.
Bài 2: Cache Manager
Tạo context manager temporarily modify cache settings.
Bài 3: Retry Context
Tạo context manager retry operations on failure.
Bài 4: Atomic File Write
Tạo context manager write to temp file, rename on success.
Bài 5: Performance Monitor
Tạo context manager track memory và CPU usage.
Tóm Tắt
✅ Context Manager: Manage resources với setup/cleanup
✅ Protocol: __enter__() và __exit__()
✅ with statement: Automatic resource management
✅ @contextmanager: Decorator tạo context managers
✅ contextlib: Utilities như suppress, redirect, ExitStack
✅ Exception handling: Return True to suppress
✅ Real-world: Transactions, locks, environment, mocking
Bài Tiếp Theo
Bài 5: Advanced Functions - *args, **kwargs, partial, closures, và function attributes! 🚀
Remember:
- Always clean up resources
- Use contextlib for simple cases
- Return False to propagate exceptions
- ExitStack for multiple resources
- Context managers = clean code! 🎯