Error Handling & Custom Exceptions

Introduction

Errors are inevitable in programming — files don't exist, network requests fail, or code receives unexpected data types. If errors aren't handled properly, your program may crash unexpectedly.

Python provides comprehensive tools to handle errors gracefully so your program can recover or fail in a controlled manner:

• Use raise to create and signal an exception
• Use try...except to handle exceptions
• Use finally for cleanup code that always runs
• Use else for code that runs only if no exception occurred
• Create custom exception classes to describe specific problems

Built-in exception types in Python include:

• Exception (base class for most exceptions)
• ValueError (inappropriate value)
• TypeError (wrong data type)
• FileNotFoundError (file doesn't exist)
• KeyError (dictionary key not found)
• IndexError (list index out of range)
• ZeroDivisionError (division by zero)

Understanding how to raise, catch, and define exceptions makes your code more robust and easier to debug.

---

Concepts

Raising Exceptions

• Use raise to create an exception
• Can raise built-in or custom exceptions
• Include descriptive error messages

raise ValueError("Age must be a positive number")

Try...Except...Else...Finally

• try: wrap risky code
• except: handle specific exceptions
• else: code that runs if no exception occurred
• finally: cleanup code that always runs

try:
    result = risky_operation()
except ValueError as e:
    print(f"Value error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")
else:
    print(f"Success: {result}")
finally:
    print("Cleanup always happens")

Exception Chaining

• Use raise...from to chain exceptions
• Preserves the original error context

try:
    process_data()
except FileNotFoundError as e:
    raise ValueError("Configuration file missing") from e

Custom Exception Classes

• Inherit from Exception or more specific base classes
• Add custom attributes and methods
• Follow naming convention: end with "Error"

class ValidationError(Exception):
    def __init__(self, message, field=None):
        super().__init__(message)
        self.field = field

class AgeValidationError(ValidationError):
    def __init__(self, age, message=None):
        self.age = age
        message = message or f"Invalid age: {age}"
        super().__init__(message, field="age")

---

Examples

Basic Error Handling

def divide_numbers(a, b):
    """Divide two numbers with error handling"""
    try:
        result = a / b
        return result
    except ZeroDivisionError:
        print("Error: Cannot divide by zero!")
        return None
    except TypeError as e:
        print(f"Error: Invalid data types - {e}")
        return None

# Usage
print(divide_numbers(10, 2))    # 5.0
print(divide_numbers(10, 0))    # Error message, returns None
print(divide_numbers("10", 2))  # Error message, returns None

File Operations with Error Handling

import json
from pathlib import Path

def read_config_file(filepath):
    """Read JSON config file with comprehensive error handling"""
    try:
        with open(filepath, 'r') as file:
            data = json.load(file)
            return data
    except FileNotFoundError:
        print(f"Error: Config file '{filepath}' not found")
        return None
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON in '{filepath}': {e}")
        return None
    except PermissionError:
        print(f"Error: Permission denied reading '{filepath}'")
        return None
    except Exception as e:
        print(f"Unexpected error reading '{filepath}': {e}")
        return None

def write_config_file(filepath, data):
    """Write JSON config file with error handling"""
    try:
        # Ensure directory exists
        Path(filepath).parent.mkdir(parents=True, exist_ok=True)
    
        with open(filepath, 'w') as file:
            json.dump(data, file, indent=2)
        print(f"Config saved to '{filepath}'")
        return True
    except PermissionError:
        print(f"Error: Permission denied writing '{filepath}'")
        return False
    except Exception as e:
        print(f"Unexpected error writing '{filepath}': {e}")
        return False

# Usage
config = read_config_file("config.json")
if config:
    print("Config loaded successfully")
else:
    # Use default config
    default_config = {"debug": False, "port": 8080}
    write_config_file("config.json", default_config)

Using Else and Finally

def process_user_data(user_id):
    """Process user data with complete error handling"""
    database_connection = None
  
    try:
        # Simulate database connection
        print(f"Connecting to database...")
        database_connection = connect_to_database()
    
        # Simulate data retrieval
        print(f"Fetching user {user_id}...")
        user_data = fetch_user(database_connection, user_id)
    
        # Validate data
        if not user_data:
            raise ValueError(f"User {user_id} not found")
    
        # Process data
        processed_data = transform_user_data(user_data)
    
    except ConnectionError as e:
        print(f"Database connection failed: {e}")
        return None
    except ValueError as e:
        print(f"Data validation error: {e}")
        return None
    except Exception as e:
        print(f"Unexpected error: {e}")
        return None
    else:
        # This runs only if no exception occurred
        print(f"Successfully processed user {user_id}")
        return processed_data
    finally:
        # This always runs
        if database_connection:
            print("Closing database connection")
            database_connection.close()

def connect_to_database():
    # Simulate database connection
    return {"connected": True}

def fetch_user(connection, user_id):
    # Simulate user fetch
    users = {1: {"name": "Alice", "email": "alice@example.com"}}
    return users.get(user_id)

def transform_user_data(data):
    # Simulate data transformation
    return {**data, "processed": True}

# Usage
result = process_user_data(1)
if result:
    print(f"Final result: {result}")

Custom Exception Hierarchy

# Base application exception
class AppError(Exception):
    """Base exception class for application errors"""
    pass

# Authentication errors
class AuthenticationError(AppError):
    """Raised when authentication fails"""
    def __init__(self, message, username=None):
        super().__init__(message)
        self.username = username

class PermissionError(AppError):
    """Raised when user lacks required permissions"""
    def __init__(self, message, required_permission=None):
        super().__init__(message)
        self.required_permission = required_permission

# Validation errors
class ValidationError(AppError):
    """Raised when data validation fails"""
    def __init__(self, message, field=None, value=None):
        super().__init__(message)
        self.field = field
        self.value = value

class EmailValidationError(ValidationError):
    """Raised when email validation fails"""
    def __init__(self, email):
        message = f"Invalid email format: {email}"
        super().__init__(message, field="email", value=email)

class PasswordValidationError(ValidationError):
    """Raised when password validation fails"""
    def __init__(self, reason):
        message = f"Password validation failed: {reason}"
        super().__init__(message, field="password")

# Usage example
class UserService:
    def create_user(self, username, email, password):
        """Create user with validation"""
        try:
            self._validate_email(email)
            self._validate_password(password)
            self._check_username_available(username)
        
            # Create user
            user = {"username": username, "email": email}
            print(f"User created: {user}")
            return user
        
        except ValidationError as e:
            print(f"Validation failed for field '{e.field}': {e}")
            raise
        except AuthenticationError as e:
            print(f"Authentication error: {e}")
            raise
  
    def _validate_email(self, email):
        if "@" not in email or "." not in email:
            raise EmailValidationError(email)
  
    def _validate_password(self, password):
        if len(password) < 8:
            raise PasswordValidationError("must be at least 8 characters")
        if not any(c.isupper() for c in password):
            raise PasswordValidationError("must contain uppercase letter")
  
    def _check_username_available(self, username):
        # Simulate checking username availability
        taken_usernames = ["admin", "root", "test"]
        if username in taken_usernames:
            raise ValidationError(
                f"Username '{username}' is already taken",
                field="username",
                value=username
            )

# Test the user service
service = UserService()

try:
    user = service.create_user("alice", "alice@example.com", "SecurePass123")
except ValidationError as e:
    print(f"Failed to create user: {e}")
except AppError as e:
    print(f"Application error: {e}")

---

Challenges

1. Safe Division Function

Create a function that safely divides two numbers and handles all possible errors.

💡 Show Answer

def safe_divide(dividend, divisor):
    """
    Safely divide two numbers with comprehensive error handling
    """
    try:
        # Validate input types
        if not isinstance(dividend, (int, float)):
            raise TypeError(f"Dividend must be a number, got {type(dividend).__name__}")
        if not isinstance(divisor, (int, float)):
            raise TypeError(f"Divisor must be a number, got {type(divisor).__name__}")
    
        # Check for zero division
        if divisor == 0:
            raise ZeroDivisionError("Cannot divide by zero")
    
        # Perform division
        result = dividend / divisor
    
        # Check for infinity or NaN
        if not (result == result):  # NaN check
            raise ValueError("Result is NaN (Not a Number)")
        if result == float('inf') or result == float('-inf'):
            raise OverflowError("Result is infinite")
    
        return result
    
    except TypeError as e:
        print(f"Type error: {e}")
        return None
    except ZeroDivisionError as e:
        print(f"Division error: {e}")
        return None
    except (ValueError, OverflowError) as e:
        print(f"Calculation error: {e}")
        return None
    except Exception as e:
        print(f"Unexpected error: {e}")
        return None

# Test cases
test_cases = [
    (10, 2),          # Normal division
    (10, 0),          # Division by zero
    ("10", 2),        # Invalid dividend type
    (10, "2"),        # Invalid divisor type
    (10, 0.0),        # Zero as float
    (float('inf'), 2), # Infinity input
    (10, float('inf')), # Division by infinity
]

for dividend, divisor in test_cases:
    print(f"safe_divide({dividend}, {divisor}) = {safe_divide(dividend, divisor)}")

# Advanced version with custom exceptions
class DivisionError(Exception):
    """Base class for division errors"""
    pass

class InvalidOperandError(DivisionError):
    """Raised when operands are invalid"""
    pass

class DivisionByZeroError(DivisionError):
    """Raised when attempting division by zero"""
    pass

def advanced_safe_divide(dividend, divisor):
    """Advanced division with custom exceptions"""
    try:
        if not isinstance(dividend, (int, float)):
            raise InvalidOperandError(f"Dividend must be numeric, got {type(dividend).__name__}")
        if not isinstance(divisor, (int, float)):
            raise InvalidOperandError(f"Divisor must be numeric, got {type(divisor).__name__}")
    
        if divisor == 0:
            raise DivisionByZeroError("Division by zero is undefined")
    
        return dividend / divisor
    
    except DivisionError:
        # Re-raise our custom errors
        raise
    except Exception as e:
        # Wrap unexpected errors
        raise DivisionError(f"Unexpected division error: {e}") from e

# Usage with custom exceptions
try:
    result = advanced_safe_divide(15, 3)
    print(f"Result: {result}")
except DivisionError as e:
    print(f"Division failed: {e}")

---

2. File Processor

Create a file processing function that handles various file-related errors.

💡 Show Answer

import json
import csv
import os
from pathlib import Path
from typing import Any, Dict, List, Optional

class FileProcessorError(Exception):
    """Base exception for file processor errors"""
    pass

class UnsupportedFileTypeError(FileProcessorError):
    """Raised when file type is not supported"""
    pass

class FileProcessor:
    """File processor with comprehensive error handling"""
  
    SUPPORTED_EXTENSIONS = {'.json', '.csv', '.txt'}
  
    def __init__(self):
        self.processed_files = []
        self.failed_files = []
  
    def process_file(self, filepath: str) -> Optional[Dict[str, Any]]:
        """
        Process a file and return its contents with metadata
        """
        filepath = Path(filepath)
    
        try:
            # Validate file existence
            if not filepath.exists():
                raise FileNotFoundError(f"File does not exist: {filepath}")
        
            # Validate file type
            if filepath.suffix.lower() not in self.SUPPORTED_EXTENSIONS:
                raise UnsupportedFileTypeError(
                    f"Unsupported file type: {filepath.suffix}. "
                    f"Supported types: {', '.join(self.SUPPORTED_EXTENSIONS)}"
                )
        
            # Check file permissions
            if not os.access(filepath, os.R_OK):
                raise PermissionError(f"No read permission for file: {filepath}")
        
            # Get file stats
            stats = filepath.stat()
            if stats.st_size == 0:
                raise ValueError(f"File is empty: {filepath}")
            if stats.st_size > 10 * 1024 * 1024:  # 10MB limit
                raise ValueError(f"File too large (>10MB): {filepath}")
        
            # Process based on file type
            content = self._read_file_content(filepath)
        
            # Create result metadata
            result = {
                'filepath': str(filepath),
                'filename': filepath.name,
                'extension': filepath.suffix,
                'size_bytes': stats.st_size,
                'content': content,
                'processed_successfully': True
            }
        
            self.processed_files.append(result)
            return result
        
        except FileNotFoundError as e:
            error_info = self._create_error_info(filepath, "File not found", str(e))
            self.failed_files.append(error_info)
            print(f"❌ File not found: {e}")
            return None
        
        except PermissionError as e:
            error_info = self._create_error_info(filepath, "Permission denied", str(e))
            self.failed_files.append(error_info)
            print(f"❌ Permission error: {e}")
            return None
        
        except UnsupportedFileTypeError as e:
            error_info = self._create_error_info(filepath, "Unsupported file type", str(e))
            self.failed_files.append(error_info)
            print(f"❌ Unsupported file: {e}")
            return None
        
        except ValueError as e:
            error_info = self._create_error_info(filepath, "Invalid file", str(e))
            self.failed_files.append(error_info)
            print(f"❌ File validation error: {e}")
            return None
        
        except json.JSONDecodeError as e:
            error_info = self._create_error_info(filepath, "JSON decode error", f"Line {e.lineno}: {e.msg}")
            self.failed_files.append(error_info)
            print(f"❌ JSON error in {filepath}: {e}")
            return None
        
        except UnicodeDecodeError as e:
            error_info = self._create_error_info(filepath, "Encoding error", str(e))
            self.failed_files.append(error_info)
            print(f"❌ Encoding error in {filepath}: {e}")
            return None
        
        except Exception as e:
            error_info = self._create_error_info(filepath, "Unexpected error", str(e))
            self.failed_files.append(error_info)
            print(f"❌ Unexpected error processing {filepath}: {e}")
            return None
  
    def _read_file_content(self, filepath: Path) -> Any:
        """Read file content based on its type"""
        extension = filepath.suffix.lower()
    
        try:
            if extension == '.json':
                with open(filepath, 'r', encoding='utf-8') as f:
                    return json.load(f)
        
            elif extension == '.csv':
                with open(filepath, 'r', encoding='utf-8') as f:
                    reader = csv.DictReader(f)
                    return list(reader)
        
            elif extension == '.txt':
                with open(filepath, 'r', encoding='utf-8') as f:
                    return f.read().strip()
        
            else:
                raise UnsupportedFileTypeError(f"No handler for {extension} files")
            
        except UnicodeDecodeError:
            # Try different encodings
            encodings = ['latin-1', 'cp1252', 'iso-8859-1']
            for encoding in encodings:
                try:
                    with open(filepath, 'r', encoding=encoding) as f:
                        if extension == '.txt':
                            return f.read().strip()
                        elif extension == '.csv':
                            reader = csv.DictReader(f)
                            return list(reader)
                except UnicodeDecodeError:
                    continue
        
            # If all encodings fail, raise the original error
            raise
  
    def _create_error_info(self, filepath: Path, error_type: str, error_message: str) -> Dict[str, Any]:
        """Create standardized error information"""
        return {
            'filepath': str(filepath),
            'filename': filepath.name if filepath else 'unknown',
            'error_type': error_type,
            'error_message': error_message,
            'processed_successfully': False
        }
  
    def process_directory(self, directory_path: str, recursive: bool = False) -> List[Dict[str, Any]]:
        """Process all supported files in a directory"""
        directory = Path(directory_path)
        results = []
    
        try:
            if not directory.exists():
                raise FileNotFoundError(f"Directory does not exist: {directory}")
        
            if not directory.is_dir():
                raise NotADirectoryError(f"Path is not a directory: {directory}")
        
            # Find files to process
            pattern = "**/*" if recursive else "*"
            files_to_process = [
                f for f in directory.glob(pattern)
                if f.is_file() and f.suffix.lower() in self.SUPPORTED_EXTENSIONS
            ]
        
            print(f"Found {len(files_to_process)} files to process in {directory}")
        
            # Process each file
            for file_path in files_to_process:
                result = self.process_file(file_path)
                if result:
                    results.append(result)
        
            return results
        
        except (FileNotFoundError, NotADirectoryError) as e:
            print(f"❌ Directory error: {e}")
            return []
        except Exception as e:
            print(f"❌ Unexpected error processing directory {directory}: {e}")
            return []
  
    def get_summary(self) -> Dict[str, Any]:
        """Get processing summary"""
        total_processed = len(self.processed_files)
        total_failed = len(self.failed_files)
        total_attempted = total_processed + total_failed
    
        success_rate = (total_processed / total_attempted * 100) if total_attempted > 0 else 0
    
        return {
            'total_attempted': total_attempted,
            'successful': total_processed,
            'failed': total_failed,
            'success_rate': f"{success_rate:.1f}%",
            'successful_files': [f['filename'] for f in self.processed_files],
            'failed_files': [f['filename'] for f in self.failed_files]
        }

# Usage example
if __name__ == "__main__":
    processor = FileProcessor()
  
    # Create some test files
    test_data = {
        "test.json": '{"name": "Alice", "age": 30}',
        "test.txt": "Hello, World!\nThis is a test file.",
        "test.csv": "name,age,city\nAlice,30,New York\nBob,25,London",
        "invalid.json": '{"invalid": json content}',  # Invalid JSON
        "empty.txt": "",  # Empty file
    }
  
    # Create test files
    for filename, content in test_data.items():
        with open(filename, 'w') as f:
            f.write(content)
  
    # Process individual files
    print("Processing individual files:")
    for filename in test_data.keys():
        result = processor.process_file(filename)
        if result:
            print(f"✅ Successfully processed {filename}")
  
    # Try non-existent file
    processor.process_file("nonexistent.txt")
  
    # Try unsupported file type
    processor.process_file("test.exe")
  
    # Print summary
    print("\nProcessing Summary:")
    summary = processor.get_summary()
    for key, value in summary.items():
        print(f"  {key}: {value}")
  
    # Clean up test files
    for filename in test_data.keys():
        try:
            os.unlink(filename)
        except FileNotFoundError:
            pass

---

3. User Input Validator

Create a comprehensive user input validation system with custom exceptions.

💡 Show Answer

import re
from datetime import datetime, date
from typing import Any, Dict, List, Optional, Union

# Custom exception hierarchy
class ValidationError(Exception):
    """Base validation exception"""
    def __init__(self, message: str, field: str = None, value: Any = None):
        super().__init__(message)
        self.field = field
        self.value = value
        self.timestamp = datetime.now()

class RequiredFieldError(ValidationError):
    """Raised when a required field is missing"""
    pass

class TypeValidationError(ValidationError):
    """Raised when field value has wrong type"""
    pass

class RangeValidationError(ValidationError):
    """Raised when field value is outside allowed range"""
    pass

class FormatValidationError(ValidationError):
    """Raised when field value has invalid format"""
    pass

class EmailValidationError(FormatValidationError):
    """Raised when email format is invalid"""
    pass

class PhoneValidationError(FormatValidationError):
    """Raised when phone format is invalid"""
    pass

class PasswordValidationError(FormatValidationError):
    """Raised when password doesn't meet requirements"""
    pass

class UserInputValidator:
    """Comprehensive user input validation system"""
  
    def __init__(self):
        self.validation_errors = []
        self.email_pattern = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
        self.phone_pattern = re.compile(r'^\+?1?-?\s?\(?[0-9]{3}\)?[-.\s]?[0-9]{3}[-.\s]?[0-9]{4}$')
  
    def validate_user_registration(self, user_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        Validate complete user registration data
        """
        self.validation_errors.clear()
        validated_data = {}
    
        try:
            # Required fields
            validated_data['username'] = self.validate_username(user_data.get('username'))
            validated_data['email'] = self.validate_email(user_data.get('email'))
            validated_data['password'] = self.validate_password(user_data.get('password'))
            validated_data['age'] = self.validate_age(user_data.get('age'))
        
            # Optional fields
            if 'first_name' in user_data:
                validated_data['first_name'] = self.validate_name(user_data['first_name'], 'first_name')
            if 'last_name' in user_data:
                validated_data['last_name'] = self.validate_name(user_data['last_name'], 'last_name')
            if 'phone' in user_data:
                validated_data['phone'] = self.validate_phone(user_data['phone'])
            if 'birth_date' in user_data:
                validated_data['birth_date'] = self.validate_birth_date(user_data['birth_date'])
        
            # If we have any validation errors, raise them
            if self.validation_errors:
                self._raise_multiple_validation_errors()
        
            print("✅ All validation passed!")
            return validated_data
        
        except ValidationError as e:
            print(f"❌ Validation failed: {e}")
            if hasattr(e, 'field') and e.field:
                print(f"   Field: {e.field}")
                print(f"   Value: {e.value}")
            raise
  
    def validate_username(self, username: Any) -> str:
        """Validate username with comprehensive checks"""
        field_name = 'username'
    
        try:
            # Check if provided
            if username is None or username == '':
                raise RequiredFieldError("Username is required", field_name, username)
        
            # Check type
            if not isinstance(username, str):
                raise TypeValidationError(
                    f"Username must be a string, got {type(username).__name__}",
                    field_name, username
                )
        
            # Clean and validate
            username = username.strip()
        
            # Length check
            if len(username) < 3:
                raise RangeValidationError(
                    "Username must be at least 3 characters long",
                    field_name, username
                )
            if len(username) > 20:
                raise RangeValidationError(
                    "Username cannot be longer than 20 characters",
                    field_name, username
                )
        
            # Format check
            if not re.match(r'^[a-zA-Z0-9_]+$', username):
                raise FormatValidationError(
                    "Username can only contain letters, numbers, and underscores",
                    field_name, username
                )
        
            # Reserved names check
            reserved_names = {'admin', 'root', 'user', 'test', 'guest', 'null', 'undefined'}
            if username.lower() in reserved_names:
                raise FormatValidationError(
                    f"Username '{username}' is reserved",
                    field_name, username
                )
        
            return username
        
        except ValidationError:
            raise
        except Exception as e:
            raise ValidationError(f"Unexpected error validating username: {e}", field_name, username)
  
    def validate_email(self, email: Any) -> str:
        """Validate email address"""
        field_name = 'email'
    
        try:
            if email is None or email == '':
                raise RequiredFieldError("Email is required", field_name, email)
        
            if not isinstance(email, str):
                raise TypeValidationError(
                    f"Email must be a string, got {type(email).__name__}",
                    field_name, email
                )
        
            email = email.strip().lower()
        
            if len(email) > 254:  # RFC 5321 limit
                raise RangeValidationError(
                    "Email address is too long (max 254 characters)",
                    field_name, email
                )
        
            if not self.email_pattern.match(email):
                raise EmailValidationError(
                    "Invalid email format",
                    field_name, email
                )
        
            # Additional checks
            local, domain = email.split('@')
            if len(local) > 64:  # RFC 5321 limit
                raise EmailValidationError(
                    "Email local part is too long (max 64 characters)",
                    field_name, email
                )
        
            # Check for common typos
            common_domains = {
                'gmial.com': 'gmail.com',
                'yahooo.com': 'yahoo.com',
                'hotmial.com': 'hotmail.com'
            }
            if domain in common_domains:
                suggested = email.replace(domain, common_domains[domain])
                raise EmailValidationError(
                    f"Did you mean '{suggested}'?",
                    field_name, email
                )
        
            return email
        
        except ValidationError:
            raise
        except Exception as e:
            raise ValidationError(f"Unexpected error validating email: {e}", field_name, email)
  
    def validate_password(self, password: Any) -> str:
        """Validate password strength"""
        field_name = 'password'
    
        try:
            if password is None or password == '':
                raise RequiredFieldError("Password is required", field_name, None)
        
            if not isinstance(password, str):
                raise TypeValidationError(
                    f"Password must be a string, got {type(password).__name__}",
                    field_name, password
                )
        
            # Length check
            if len(password) < 8:
                raise PasswordValidationError(
                    "Password must be at least 8 characters long",
                    field_name, len(password)
                )
            if len(password) > 128:
                raise PasswordValidationError(
                    "Password is too long (max 128 characters)",
                    field_name, len(password)
                )
        
            # Character type checks
            has_upper = any(c.isupper() for c in password)
            has_lower = any(c.islower() for c in password)
            has_digit = any(c.isdigit() for c in password)
            has_special = any(c in "!@#$%^&*()_+-=[]{}|;:,.<>?" for c in password)
        
            missing_requirements = []
            if not has_upper:
                missing_requirements.append("uppercase letter")
            if not has_lower:
                missing_requirements.append("lowercase letter")
            if not has_digit:
                missing_requirements.append("digit")
            if not has_special:
                missing_requirements.append("special character")
        
            if missing_requirements:
                raise PasswordValidationError(
                    f"Password must contain: {', '.join(missing_requirements)}",
                    field_name, "***"
                )
        
            # Common password check
            common_passwords = {
                'password', 'password123', '123456789', 'qwerty123',
                'admin123', 'letmein', 'welcome123', 'password1'
            }
            if password.lower() in common_passwords:
                raise PasswordValidationError(
                    "Password is too common, please choose a stronger password",
                    field_name, "***"
                )
        
            # Don't return the actual password for security
            return password
        
        except ValidationError:
            raise
        except Exception as e:
            raise ValidationError(f"Unexpected error validating password: {e}", field_name, "***")
  
    def validate_age(self, age: Any) -> int:
        """Validate age"""
        field_name = 'age'
    
        try:
            if age is None:
                raise RequiredFieldError("Age is required", field_name, age)
        
            # Convert to int if possible
            try:
                age = int(age)
            except (TypeError, ValueError):
                raise TypeValidationError(
                    f"Age must be a number, got {type(age).__name__}",
                    field_name, age
                )
        
            # Range check
            if age < 13:
                raise RangeValidationError(
                    "Users must be at least 13 years old",
                    field_name, age
                )
            if age > 120:
                raise RangeValidationError(
                    "Age cannot be greater than 120",
                    field_name, age
                )
        
            return age
        
        except ValidationError:
            raise
        except Exception as e:
            raise ValidationError(f"Unexpected error validating age: {e}", field_name, age)
  
    def validate_name(self, name: Any, field_name: str) -> str:
        """Validate first/last name"""
        try:
            if not name or name == '':
                return ''  # Names are optional
        
            if not isinstance(name, str):
                raise TypeValidationError(
                    f"{field_name.replace('_', ' ').title()} must be a string",
                    field_name, name
                )
        
            name = name.strip()
        
            if len(name) > 50:
                raise RangeValidationError(
                    f"{field_name.replace('_', ' ').title()} is too long (max 50 characters)",
                    field_name, name
                )
        
            if not re.match(r"^[a-zA-Z\s\-'\.]+$", name):
                raise FormatValidationError(
                    f"{field_name.replace('_', ' ').title()} contains invalid characters",
                    field_name, name
                )
        
            return name.title()  # Proper case
        
        except ValidationError:
            raise
        except Exception as e:
            raise ValidationError(f"Unexpected error validating {field_name}: {e}", field_name, name)
  
    def validate_phone(self, phone: Any) -> str:
        """Validate phone number"""
        field_name = 'phone'
    
        try:
            if not phone or phone == '':
                return ''  # Phone is optional
        
            if not isinstance(phone, str):
                raise TypeValidationError(
                    "Phone number must be a string",
                    field_name, phone
                )
        
            # Clean phone number
            phone = re.sub(r'[\s\-\(\)\.]+', '', phone)
        
            if not self.phone_pattern.match(phone):
                raise PhoneValidationError(
                    "Invalid phone number format. Use format: (123) 456-7890",
                    field_name, phone
                )
        
            return phone
        
        except ValidationError:
            raise
        except Exception as e:
            raise ValidationError(f"Unexpected error validating phone: {e}", field_name, phone)
  
    def validate_birth_date(self, birth_date: Any) -> date:
        """Validate birth date"""
        field_name = 'birth_date'
    
        try:
            if not birth_date:
                return None  # Birth date is optional
        
            # Handle different input types
            if isinstance(birth_date, str):
                try:
                    # Try different date formats
                    for fmt in ['%Y-%m-%d', '%m/%d/%Y', '%d/%m/%Y']:
                        try:
                            birth_date = datetime.strptime(birth_date, fmt).date()
                            break
                        except ValueError:
                            continue
                    else:
                        raise FormatValidationError(
                            "Invalid date format. Use YYYY-MM-DD",
                            field_name, birth_date
                        )
                except Exception:
                    raise FormatValidationError(
                        "Invalid date format. Use YYYY-MM-DD",
                        field_name, birth_date
                    )
            elif isinstance(birth_date, datetime):
                birth_date = birth_date.date()
            elif not isinstance(birth_date, date):
                raise TypeValidationError(
                    "Birth date must be a date string or date object",
                    field_name, birth_date
                )
        
            # Validate date range
            today = date.today()
            if birth_date > today:
                raise RangeValidationError(
                    "Birth date cannot be in the future",
                    field_name, birth_date
                )
        
            min_date = date(today.year - 120, today.month, today.day)
            if birth_date < min_date:
                raise RangeValidationError(
                    "Birth date is too far in the past",
                    field_name, birth_date
                )
        
            return birth_date
        
        except ValidationError:
            raise
        except Exception as e:
            raise ValidationError(f"Unexpected error validating birth date: {e}", field_name, birth_date)
  
    def _raise_multiple_validation_errors(self):
        """Raise multiple validation errors as a single exception"""
        error_messages = []
        for error in self.validation_errors:
            if hasattr(error, 'field') and error.field:
                error_messages.append(f"{error.field}: {error}")
            else:
                error_messages.append(str(error))
    
        raise ValidationError(f"Multiple validation errors: {'; '.join(error_messages)}")

# Test the validator
def test_user_validator():
    """Test the user input validator"""
    validator = UserInputValidator()
  
    # Test valid user data
    print("Testing valid user data:")
    valid_user = {
        'username': 'alice_doe',
        'email': 'alice@example.com',
        'password': 'SecurePass123!',
        'age': 25,
        'first_name': 'Alice',
        'last_name': 'Doe',
        'phone': '(555) 123-4567',
        'birth_date': '1998-01-15'
    }
  
    try:
        result = validator.validate_user_registration(valid_user)
        print("Validated user data:", result)
    except ValidationError as e:
        print(f"Validation failed: {e}")
  
    # Test invalid user data
    print("\nTesting invalid user data:")
    invalid_users = [
        {
            'username': 'ab',  # Too short
            'email': 'invalid-email',
            'password': '123',  # Too weak
            'age': 5  # Too young
        },
        {
            'username': 'admin',  # Reserved name
            'email': 'user@gmial.com',  # Typo in domain
            'password': 'password',  # Too common
            'age': 'not_a_number'  # Wrong type
        }
    ]
  
    for i, invalid_user in enumerate(invalid_users, 1):
        print(f"\nTest case {i}:")
        try:
            validator.validate_user_registration(invalid_user)
        except ValidationError as e:
            print(f"Expected validation error: {e}")

if __name__ == "__main__":
    test_user_validator()

---

4. API Error Handler

Create a decorator for handling API errors with retry logic.

💡 Show Answer

import functools
import time
import random
from typing import Callable, Any, Type, Tuple, Optional
from datetime import datetime

# Custom API exceptions
class APIError(Exception):
    """Base API exception"""
    def __init__(self, message: str, status_code: int = None):
        super().__init__(message)
        self.status_code = status_code
        self.timestamp = datetime.now()

class NetworkError(APIError):
    """Network-related API errors"""
    pass

class AuthenticationError(APIError):
    """Authentication failures"""
    pass

class RateLimitError(APIError):
    """Rate limiting errors"""
    pass

class ServerError(APIError):
    """Server-side errors"""
    pass

class ClientError(APIError):
    """Client-side errors"""
    pass

def api_error_handler(
    max_retries: int = 3,
    retry_delay: float = 1.0,
    exponential_backoff: bool = True,
    retry_on: Tuple[Type[Exception], ...] = (NetworkError, ServerError, RateLimitError),
    circuit_breaker_threshold: int = 5
):
    """
    Decorator for handling API errors with retry logic and circuit breaker
  
    Args:
        max_retries: Maximum number of retry attempts
        retry_delay: Initial delay between retries in seconds
        exponential_backoff: Whether to use exponential backoff
        retry_on: Exception types that should trigger a retry
        circuit_breaker_threshold: Number of failures before circuit breaker opens
    """
  
    # Circuit breaker state (shared across all decorated functions)
    _circuit_breaker = {
        'failure_count': 0,
        'last_failure_time': None,
        'is_open': False,
        'recovery_time': 60  # seconds
    }
  
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            # Check circuit breaker
            if _circuit_breaker['is_open']:
                if (datetime.now().timestamp() - _circuit_breaker['last_failure_time']) > _circuit_breaker['recovery_time']:
                    # Try to close circuit breaker
                    _circuit_breaker['is_open'] = False
                    _circuit_breaker['failure_count'] = 0
                    print("🔄 Circuit breaker: Attempting to close")
                else:
                    raise APIError("Circuit breaker is open - service temporarily unavailable")
        
            last_exception = None
        
            for attempt in range(max_retries + 1):
                try:
                    print(f"🔄 Attempt {attempt + 1}/{max_retries + 1} for {func.__name__}")
                
                    # Call the original function
                    result = func(*args, **kwargs)
                
                    # Success - reset circuit breaker
                    if _circuit_breaker['failure_count'] > 0:
                        print("✅ Circuit breaker: Reset after successful call")
                        _circuit_breaker['failure_count'] = 0
                        _circuit_breaker['is_open'] = False
                
                    return result
                
                except retry_on as e:
                    last_exception = e
                    print(f"❌ Attempt {attempt + 1} failed: {e}")
                
                    # Update circuit breaker
                    _circuit_breaker['failure_count'] += 1
                    _circuit_breaker['last_failure_time'] = datetime.now().timestamp()
                
                    if _circuit_breaker['failure_count'] >= circuit_breaker_threshold:
                        _circuit_breaker['is_open'] = True
                        print(f"⚡ Circuit breaker opened after {_circuit_breaker['failure_count']} failures")
                        raise APIError("Circuit breaker opened due to repeated failures") from e
                
                    # Don't retry on the last attempt
                    if attempt == max_retries:
                        break
                
                    # Calculate delay with optional exponential backoff
                    if exponential_backoff:
                        delay = retry_delay * (2 ** attempt) + random.uniform(0, 1)
                    else:
                        delay = retry_delay + random.uniform(0, 0.5)  # Add jitter
                
                    print(f"⏳ Waiting {delay:.2f}s before retry...")
                    time.sleep(delay)
                
                except (AuthenticationError, ClientError) as e:
                    # Don't retry these errors
                    print(f"❌ Non-retryable error: {e}")
                    raise
                
                except Exception as e:
                    # Unexpected error
                    print(f"❌ Unexpected error: {e}")
                    raise APIError(f"Unexpected error in {func.__name__}: {e}") from e
        
            # All retries exhausted
            print(f"❌ All {max_retries + 1} attempts failed for {func.__name__}")
            raise APIError(f"Failed after {max_retries + 1} attempts") from last_exception
    
        return wrapper
    return decorator

# Simulate various API functions for testing
class MockAPIClient:
    """Mock API client for demonstration"""
  
    def __init__(self):
        self.call_count = 0
        self.auth_token = None
  
    def authenticate(self, username: str, password: str) -> str:
        """Simulate authentication"""
        if username == "admin" and password == "secret123":
            self.auth_token = "mock_token_123"
            return self.auth_token
        else:
            raise AuthenticationError("Invalid credentials", 401)
  
    @api_error_handler(max_retries=3, retry_delay=0.5)
    def get_user_data(self, user_id: int) -> dict:
        """Simulate getting user data with potential failures"""
        self.call_count += 1
    
        if not self.auth_token:
            raise AuthenticationError("Authentication required", 401)
    
        # Simulate various failure scenarios
        if self.call_count <= 2:  # Fail first 2 attempts
            failure_type = random.choice([
                NetworkError("Connection timeout", 503),
                ServerError("Internal server error", 500),
                RateLimitError("Rate limit exceeded", 429)
            ])
            raise failure_type
    
        # Success on 3rd attempt
        return {
            "id": user_id,
            "name": f"User {user_id}",
            "email": f"user{user_id}@example.com",
            "call_count": self.call_count
        }
  
    @api_error_handler(max_retries=2, exponential_backoff=True)
    def create_user(self, user_data: dict) -> dict:
        """Simulate creating a user"""
        if not self.auth_token:
            raise AuthenticationError("Authentication required", 401)
    
        # Validate input
        if not user_data.get("name"):
            raise ClientError("Name is required", 400)
    
        if not user_data.get("email"):
            raise ClientError("Email is required", 400)
    
        # Simulate server error on first attempt
        if random.random() < 0.7:  # 70% chance of failure
            raise ServerError("Database temporarily unavailable", 503)
    
        return {
            "id": random.randint(1000, 9999),
            "created": True,
            **user_data
        }
  
    @api_error_handler(max_retries=1, retry_on=(NetworkError,))
    def delete_user(self, user_id: int) -> bool:
        """Simulate deleting a user - only retry network errors"""
        if not self.auth_token:
            raise AuthenticationError("Authentication required", 401)
    
        if user_id == 999:  # Special case for testing
            raise ClientError("User cannot be deleted", 403)
    
        # Simulate network issues
        if random.random() < 0.3:  # 30% chance
            raise NetworkError("Connection lost", 503)
    
        return True

# Advanced decorator with logging and metrics
class APIMetrics:
    """Track API call metrics"""
    def __init__(self):
        self.calls = {}
  
    def record_call(self, func_name: str, success: bool, attempts: int, duration: float):
        if func_name not in self.calls:
            self.calls[func_name] = {
                'total_calls': 0,
                'successful_calls': 0,
                'failed_calls': 0,
                'total_attempts': 0,
                'total_duration': 0.0,
                'avg_duration': 0.0,
                'success_rate': 0.0
            }
    
        stats = self.calls[func_name]
        stats['total_calls'] += 1
        stats['total_attempts'] += attempts
        stats['total_duration'] += duration
    
        if success:
            stats['successful_calls'] += 1
        else:
            stats['failed_calls'] += 1
    
        stats['avg_duration'] = stats['total_duration'] / stats['total_calls']
        stats['success_rate'] = (stats['successful_calls'] / stats['total_calls']) * 100
  
    def get_report(self) -> dict:
        return self.calls

# Global metrics instance
api_metrics = APIMetrics()

def advanced_api_error_handler(
    max_retries: int = 3,
    retry_delay: float = 1.0,
    timeout: float = 30.0,
    log_calls: bool = True
):
    """Advanced API error handler with metrics and logging"""
  
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            start_time = time.time()
            attempts = 0
            success = False
        
            try:
                for attempt in range(max_retries + 1):
                    attempts += 1
                    attempt_start = time.time()
                
                    try:
                        if log_calls:
                            print(f"📞 {func.__name__} - Attempt {attempt + 1}")
                    
                        # Add timeout to function call
                        result = func(*args, **kwargs)
                    
                        success = True
                        duration = time.time() - start_time
                    
                        if log_calls:
                            print(f"✅ {func.__name__} succeeded after {attempts} attempts in {duration:.2f}s")
                    
                        api_metrics.record_call(func.__name__, success, attempts, duration)
                        return result
                    
                    except (NetworkError, ServerError, RateLimitError) as e:
                        attempt_duration = time.time() - attempt_start
                    
                        if log_calls:
                            print(f"❌ {func.__name__} attempt {attempt + 1} failed: {e} (took {attempt_duration:.2f}s)")
                    
                        if attempt == max_retries:
                            break
                    
                        # Exponential backoff with jitter
                        delay = min(retry_delay * (2 ** attempt), 30) + random.uniform(0, 1)
                        time.sleep(delay)
                    
                    except Exception as e:
                        # Non-retryable error
                        if log_calls:
                            print(f"❌ {func.__name__} failed with non-retryable error: {e}")
                        raise
            
                # All attempts failed
                duration = time.time() - start_time
                api_metrics.record_call(func.__name__, success, attempts, duration)
                raise APIError(f"{func.__name__} failed after {attempts} attempts")
            
            except Exception as e:
                duration = time.time() - start_time
                api_metrics.record_call(func.__name__, success, attempts, duration)
                raise
    
        return wrapper
    return decorator

# Test the error handling system
def test_api_error_handling():
    """Test the API error handling system"""
    print("Testing API Error Handling System")
    print("=" * 50)
  
    client = MockAPIClient()
  
    # Test authentication (should succeed)
    try:
        token = client.authenticate("admin", "secret123")
        print(f"✅ Authentication successful: {token}")
    except AuthenticationError as e:
        print(f"❌ Authentication failed: {e}")
  
    # Test user data retrieval (will retry and eventually succeed)
    try:
        user_data = client.get_user_data(123)
        print(f"✅ User data retrieved: {user_data}")
    except APIError as e:
        print(f"❌ Failed to get user data: {e}")
  
    # Test user creation (might fail)
    try:
        new_user = client.create_user({
            "name": "Alice Johnson",
            "email": "alice@example.com"
        })
        print(f"✅ User created: {new_user}")
    except APIError as e:
        print(f"❌ Failed to create user: {e}")
  
    # Test user deletion
    try:
        result = client.delete_user(456)
        print(f"✅ User deleted: {result}")
    except APIError as e:
        print(f"❌ Failed to delete user: {e}")
  
    # Print metrics
    print("\nAPI Metrics Report:")
    print("-" * 20)
    report = api_metrics.get_report()
    for func_name, stats in report.items():
        print(f"\n{func_name}:")
        for key, value in stats.items():
            if isinstance(value, float):
                print(f"  {key}: {value:.2f}")
            else:
                print(f"  {key}: {value}")

if __name__ == "__main__":
    test_api_error_handling()