Bài 14: Exception Handling - Xử Lý Lỗi (Phần 2)
Mục Tiêu Bài Học
Sau khi hoàn thành bài này, bạn sẽ:
- ✅ Tạo custom exception classes
- ✅ Sử dụng exception chaining
- ✅ Hiểu và dùng context managers
- ✅ Làm việc với traceback
- ✅ Apply advanced error handling patterns
Custom Exceptions
Tạo exception classes riêng cho application của bạn.
Basic Custom Exception
# Simple custom exceptionclass CustomError(Exception): """Base exception for custom errors.""" pass # Raise custom exceptiondef check_value(value): if value < 0: raise CustomError("Value cannot be negative") return value try: check_value(-5)except CustomError as e: print(f"Custom error: {e}") # Exception with custom messageclass ValidationError(Exception): """Validation failed.""" def __init__(self, field, message): self.field = field self.message = message super().__init__(f"{field}: {message}") # Usagetry: raise ValidationError("email", "Invalid email format")except ValidationError as e: print(f"Field: {e.field}") print(f"Message: {e.message}")
Exception Hierarchy
# Create exception hierarchyclass ApplicationError(Exception): """Base exception for application.""" pass class DatabaseError(ApplicationError): """Database-related errors.""" pass class ConnectionError(DatabaseError): """Database connection failed.""" pass class QueryError(DatabaseError): """Query execution failed.""" pass class ValidationError(ApplicationError): """Validation errors.""" pass class AuthenticationError(ApplicationError): """Authentication errors.""" pass # Usagedef connect_db(): raise ConnectionError("Cannot connect to database") def execute_query(query): raise QueryError(f"Invalid query: {query}") # Catch hierarchytry: connect_db()except ConnectionError as e: print(f"Connection error: {e}")except DatabaseError as e: print(f"Database error: {e}")except ApplicationError as e: print(f"Application error: {e}")
Rich Custom Exceptions
# Exception with attributesclass InsufficientFundsError(Exception): """Insufficient funds for transaction.""" def __init__(self, balance, amount): self.balance = balance self.amount = amount self.deficit = amount - balance message = ( f"Insufficient funds: " f"Balance ${balance:.2f}, " f"Required ${amount:.2f}, " f"Short ${self.deficit:.2f}" ) super().__init__(message) # Usagedef withdraw(balance, amount): if amount > balance: raise InsufficientFundsError(balance, amount) return balance - amount try: balance = withdraw(100, 150)except InsufficientFundsError as e: print(e) print(f"Deficit: ${e.deficit:.2f}") # Exception with contextclass APIError(Exception): """API request failed.""" def __init__(self, endpoint, status_code, message): self.endpoint = endpoint self.status_code = status_code self.message = message super().__init__( f"API Error at {endpoint}: " f"[{status_code}] {message}" ) def is_client_error(self): """Check if 4xx error.""" return 400 <= self.status_code < 500 def is_server_error(self): """Check if 5xx error.""" return 500 <= self.status_code < 600 # Usagetry: raise APIError("/users", 404, "User not found")except APIError as e: print(e) if e.is_client_error(): print("Client error - check request") elif e.is_server_error(): print("Server error - try again later")
Exception Chaining
Link exceptions để preserve error context.
from Keyword
# Exception chaining with 'from'def process_data(data): try: result = int(data) return result except ValueError as e: # Chain exceptions raise TypeError("Data must be numeric") from e # Usagetry: process_data("abc")except TypeError as e: print(f"Error: {e}") print(f"Caused by: {e.__cause__}") print(f"Cause type: {type(e.__cause__).__name__}") # Output:# Error: Data must be numeric# Caused by: invalid literal for int() with base 10: 'abc'# Cause type: ValueError # Real-world exampledef load_config(filename): try: with open(filename, 'r') as f: import json return json.load(f) except FileNotFoundError as e: raise ConfigError(f"Config file not found: {filename}") from e except json.JSONDecodeError as e: raise ConfigError(f"Invalid JSON in config file") from e class ConfigError(Exception): """Configuration error.""" pass
Suppress Exception Context
# Suppress original exception with 'from None'def validate_age(age): try: age = int(age) if age < 0: raise ValueError("Age cannot be negative") return age except ValueError: # Suppress chaining raise ValueError("Invalid age provided") from None # Without 'from None' - shows both exceptionstry: validate_age("abc")except ValueError as e: print(e)# Shows: invalid literal... AND Invalid age provided # With 'from None' - shows only new exceptiontry: validate_age("abc")except ValueError as e: print(e)# Shows only: Invalid age provided
Exception Context
# Implicit exception chaining (during handling)def risky_operation(): try: result = 10 / 0 except ZeroDivisionError: # New exception during handling raise ValueError("Operation failed") try: risky_operation()except ValueError as e: print(f"Error: {e}") print(f"Context: {e.__context__}") # Original ZeroDivisionError # Check exception chaindef print_exception_chain(exception): """Print full exception chain.""" print(f"Exception: {exception}") if exception.__cause__: print(f" Caused by: {exception.__cause__}") if exception.__context__ and exception.__context__ is not exception.__cause__: print(f" During handling of: {exception.__context__}") try: process_data("invalid")except Exception as e: print_exception_chain(e)
Context Managers
Context managers handle setup và cleanup automatically.
with Statement
# Basic with statementwith open('data.txt', 'r') as file: content = file.read()# File automatically closed # 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 # Nested with (Python 3.10+)with ( open('input.txt', 'r') as infile, open('output.txt', 'w') as outfile): content = infile.read() outfile.write(content)
Creating Context Managers
# Class-based context managerclass FileHandler: """Context manager for file operations.""" def __init__(self, filename, mode): self.filename = filename self.mode = mode self.file = None def __enter__(self): """Setup - called when entering with block.""" print(f"Opening {self.filename}") self.file = open(self.filename, self.mode) return self.file def __exit__(self, exc_type, exc_value, traceback): """Cleanup - called when exiting with block.""" print(f"Closing {self.filename}") if self.file: self.file.close() # Return False to propagate exceptions # Return True to suppress exceptions if exc_type is not None: print(f"Exception occurred: {exc_type.__name__}") return False # Don't suppress exceptions # Usagewith FileHandler('data.txt', 'r') as file: content = file.read() print(content) # Function-based context managerfrom contextlib import contextmanager @contextmanagerdef file_handler(filename, mode): """Context manager using generator.""" print(f"Opening {filename}") file = open(filename, mode) try: yield file # Provide resource finally: print(f"Closing {filename}") file.close() # Usagewith file_handler('data.txt', 'r') as file: content = file.read()
Practical Context Managers
from contextlib import contextmanagerimport time # Timer context manager@contextmanagerdef timer(name): """Measure execution time.""" print(f"Starting {name}...") start = time.time() try: yield finally: elapsed = time.time() - start print(f"{name} took {elapsed:.2f} seconds") # Usagewith timer("Data processing"): # Do some work time.sleep(1) process_data() # Database transaction context manager@contextmanagerdef transaction(connection): """Database transaction.""" try: yield connection connection.commit() print("Transaction committed") except Exception as e: connection.rollback() print(f"Transaction rolled back: {e}") raise # Temporary directory@contextmanagerdef temp_directory(): """Create and cleanup temporary directory.""" import tempfile import shutil temp_dir = tempfile.mkdtemp() print(f"Created temp dir: {temp_dir}") try: yield temp_dir finally: shutil.rmtree(temp_dir) print(f"Cleaned up temp dir: {temp_dir}") # Usagewith temp_directory() as tmpdir: # Use temporary directory filepath = os.path.join(tmpdir, 'temp_file.txt') with open(filepath, 'w') as f: f.write("Temporary data")
Traceback
Xem và xử lý stack traces.
Print Traceback
import traceback # Print exception tracebacktry: result = 10 / 0except ZeroDivisionError: print("Exception occurred!") traceback.print_exc() # Capture traceback as stringtry: result = 10 / 0except ZeroDivisionError: error_trace = traceback.format_exc() print("Captured traceback:") print(error_trace) # Log to file with open('error.log', 'a') as f: f.write(error_trace) # Get traceback infotry: result = 10 / 0except ZeroDivisionError as e: tb = traceback.extract_tb(e.__traceback__) for frame in tb: print(f"File: {frame.filename}") print(f"Line: {frame.lineno}") print(f"Function: {frame.name}") print(f"Code: {frame.line}")
Format Traceback
import tracebackimport sys def format_exception_info(): """Format exception information.""" exc_type, exc_value, exc_traceback = sys.exc_info() if exc_type is None: return "No exception" # Format traceback tb_lines = traceback.format_exception(exc_type, exc_value, exc_traceback) return ''.join(tb_lines) # Usagetry: # Some code that fails x = 1 / 0except: error_info = format_exception_info() print(error_info) # Save to log with open('error.log', 'a') as f: f.write(f"\n{'='*50}\n") f.write(error_info)
Error Logging
Log errors instead of just printing.
Basic Logging
import logging # Setup logginglogging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', filename='app.log') # Log exceptionstry: result = 10 / 0except ZeroDivisionError as e: logging.error(f"Division error: {e}") logging.exception("Exception occurred") # Includes traceback # Different log levelslogging.debug("Debug message")logging.info("Info message")logging.warning("Warning message")logging.error("Error message")logging.critical("Critical message")
Advanced Logging
import loggingfrom datetime import datetime # Custom loggerclass ErrorLogger: """Custom error logger.""" def __init__(self, log_file='errors.log'): self.logger = logging.getLogger(__name__) self.logger.setLevel(logging.ERROR) # File handler handler = logging.FileHandler(log_file) formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) handler.setFormatter(formatter) self.logger.addHandler(handler) def log_error(self, error, context=None): """Log error with context.""" message = f"Error: {error}" if context: message += f" | Context: {context}" self.logger.error(message) def log_exception(self, exception, context=None): """Log exception with traceback.""" message = f"Exception: {exception}" if context: message += f" | Context: {context}" self.logger.exception(message) # Usagelogger = ErrorLogger() try: result = risky_operation()except Exception as e: logger.log_exception(e, context={"user_id": 123, "action": "process_data"})
Ví Dụ Thực Tế
1. API Client with Custom Exceptions
class APIException(Exception): """Base API exception.""" pass class APIConnectionError(APIException): """Connection failed.""" pass class APIAuthenticationError(APIException): """Authentication failed.""" pass class APIRateLimitError(APIException): """Rate limit exceeded.""" def __init__(self, retry_after): self.retry_after = retry_after super().__init__(f"Rate limit exceeded. Retry after {retry_after} seconds") class APIClient: """API client with error handling.""" def __init__(self, api_key): self.api_key = api_key def request(self, endpoint): """Make API request.""" try: # Simulate request if not self.api_key: raise APIAuthenticationError("API key required") if endpoint == "/rate-limited": raise APIRateLimitError(retry_after=60) if endpoint == "/error": raise APIConnectionError("Connection timeout") return {"status": "success", "data": []} except APIAuthenticationError: print("Please provide valid API key") raise except APIRateLimitError as e: print(f"Rate limited. Retry after {e.retry_after}s") raise except APIException as e: print(f"API error: {e}") raise # Usageclient = APIClient("my-api-key") try: result = client.request("/rate-limited")except APIRateLimitError as e: print(f"Waiting {e.retry_after} seconds...")except APIException as e: print(f"API failed: {e}")
2. Database Manager with Transaction
from contextlib import contextmanager class DatabaseError(Exception): """Database error.""" pass class TransactionError(DatabaseError): """Transaction error.""" pass class Database: """Database with transaction support.""" def __init__(self): self.in_transaction = False self.changes = [] def execute(self, query): """Execute query.""" print(f"Executing: {query}") self.changes.append(query) def commit(self): """Commit transaction.""" if not self.in_transaction: raise TransactionError("No active transaction") print("Committing changes...") self.in_transaction = False self.changes = [] def rollback(self): """Rollback transaction.""" if not self.in_transaction: raise TransactionError("No active transaction") print("Rolling back changes...") self.in_transaction = False self.changes = [] @contextmanager def transaction(self): """Transaction context manager.""" self.in_transaction = True print("Starting transaction...") try: yield self self.commit() except Exception as e: self.rollback() raise TransactionError("Transaction failed") from e # Usagedb = Database() try: with db.transaction(): db.execute("INSERT INTO users (name) VALUES ('Alice')") db.execute("INSERT INTO users (name) VALUES ('Bob')") # All changes committedexcept TransactionError as e: print(f"Transaction failed: {e}") try: with db.transaction(): db.execute("INSERT INTO users (name) VALUES ('Charlie')") raise ValueError("Something went wrong!") db.execute("INSERT INTO users (name) VALUES ('David')")except TransactionError as e: print(f"Transaction failed: {e}") print(f"Caused by: {e.__cause__}")
3. Validation Framework
class ValidationError(Exception): """Validation error.""" def __init__(self, errors): self.errors = errors messages = [f"{field}: {msg}" for field, msg in errors.items()] super().__init__("; ".join(messages)) class Validator: """Data validator.""" @staticmethod def validate_user(data): """Validate user data.""" errors = {} # Validate username if 'username' not in data: errors['username'] = "Required field" elif len(data['username']) < 3: errors['username'] = "Must be at least 3 characters" # Validate email if 'email' not in data: errors['email'] = "Required field" elif '@' not in data['email']: errors['email'] = "Invalid email format" # Validate age if 'age' in data: try: age = int(data['age']) if age < 0 or age > 150: errors['age'] = "Must be between 0 and 150" except ValueError: errors['age'] = "Must be a number" # Raise if errors if errors: raise ValidationError(errors) return True # Usagedef create_user(data): """Create user with validation.""" try: Validator.validate_user(data) print(f"User {data['username']} created successfully!") return True except ValidationError as e: print("Validation failed:") for field, message in e.errors.items(): print(f" - {field}: {message}") return False # Testcreate_user({ "username": "ab", "email": "invalid", "age": "not a number"})# Shows all validation errors at once
4. Retry Decorator with Logging
import timeimport loggingfrom functools import wraps logging.basicConfig(level=logging.INFO) class RetryError(Exception): """Retry failed.""" pass def retry(max_attempts=3, delay=1, exceptions=(Exception,)): """ Retry decorator. Args: max_attempts: Maximum retry attempts delay: Delay between retries (seconds) exceptions: Tuple of exceptions to catch """ def decorator(func): @wraps(func) def wrapper(*args, **kwargs): last_exception = None for attempt in range(1, max_attempts + 1): try: logging.info(f"Attempt {attempt}/{max_attempts}: {func.__name__}") return func(*args, **kwargs) except exceptions as e: last_exception = e logging.warning( f"Attempt {attempt} failed: {e}" ) if attempt < max_attempts: logging.info(f"Retrying in {delay}s...") time.sleep(delay) else: logging.error(f"All {max_attempts} attempts failed") # All attempts failed raise RetryError( f"Failed after {max_attempts} attempts" ) from last_exception return wrapper return decorator # Usage@retry(max_attempts=3, delay=1, exceptions=(ValueError, ConnectionError))def unstable_operation(): """Simulated unstable operation.""" import random if random.random() < 0.7: # 70% failure rate raise ValueError("Random failure") return "Success!" try: result = unstable_operation() print(f"Result: {result}")except RetryError as e: print(f"Operation failed: {e}") print(f"Original error: {e.__cause__}")
5. Resource Manager
from contextlib import contextmanager class ResourceError(Exception): """Resource management error.""" pass class ResourceManager: """Manage multiple resources.""" def __init__(self): self.resources = [] def acquire(self, resource): """Acquire resource.""" print(f"Acquiring: {resource}") self.resources.append(resource) return resource def release_all(self): """Release all resources.""" print("Releasing all resources...") while self.resources: resource = self.resources.pop() print(f"Releasing: {resource}") @contextmanager def managed_resources(self, *resource_names): """Context manager for resources.""" try: # Acquire all resources resources = [self.acquire(name) for name in resource_names] yield resources except Exception as e: print(f"Error occurred: {e}") raise ResourceError("Resource operation failed") from e finally: # Always cleanup self.release_all() # Usagemanager = ResourceManager() try: with manager.managed_resources("Database", "FileHandle", "NetworkSocket") as resources: print(f"Using resources: {resources}") # Do work with resources raise ValueError("Something went wrong!")except ResourceError as e: print(f"Resource error: {e}")# Resources are cleaned up even if error occurs
Best Practices Summary
# 1. Create exception hierarchyclass AppError(Exception): """Base exception.""" pass class DataError(AppError): """Data-related errors.""" pass # 2. Use exception chainingtry: operation()except ValueError as e: raise CustomError("Operation failed") from e # 3. Use context managers for cleanupwith resource_manager() as resource: use_resource(resource) # 4. Log exceptions properlyimport loggingtry: risky_operation()except Exception: logging.exception("Operation failed") # 5. Provide context in exceptionsraise ValueError( f"Invalid value {value}: " f"Expected range {min_val}-{max_val}") # 6. Don't catch and ignore# ❌ Badtry: operation()except Exception: pass # Silent failure! # ✅ Goodtry: operation()except SpecificError as e: logging.error(f"Expected error: {e}") handle_error(e)
Bài Tập Thực Hành
Bài 1: Custom Exception Hierarchy
Tạo exception hierarchy cho e-commerce:
EcommerceError(base)ProductError,OrderError,PaymentError- Each với specific subtypes
Bài 2: Context Manager
Viết context manager DatabaseConnection:
- Open connection on enter
- Auto commit on success
- Rollback on error
- Always close connection
Bài 3: Retry with Backoff
Viết decorator retry_with_backoff:
- Exponential backoff (1s, 2s, 4s, 8s)
- Max attempts configurable
- Log each attempt
- Support specific exceptions
Bài 4: Error Reporter
Tạo ErrorReporter class:
- Capture exception details
- Format traceback
- Save to log file
- Send email notification (mock)
Bài 5: Validation Framework
Tạo validation framework:
- Multiple validators
- Collect all errors
- Custom error messages
- Context manager for validation
Tóm Tắt
✅ Custom exceptions: Tạo exception classes riêng
✅ Exception hierarchy: Organize exceptions logically
✅ Exception chaining: raise ... from ... preserve context
✅ Context managers: __enter__, __exit__, @contextmanager
✅ Traceback: traceback.print_exc(), format_exc()
✅ Logging: logging.exception() includes traceback
✅ Best practice: Specific exceptions, proper cleanup, good error messages
Bài Tiếp Theo
Remember:
- Create custom exception hierarchy for your app
- Use exception chaining to preserve context
- Context managers ensure cleanup
- Log exceptions with full traceback
- Provide helpful error messages with context!