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

Bài 15: Object-Oriented Programming (OOP) - Classes, objects, inheritance, polymorphism, encapsulation.

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!

#traceback

Exception Handling - Xử Lý Lỗi (Phần 2)