Error Handling Pro

Welcome to professional error handling! In this final lesson, you'll build a comprehensive error management system that provides clear feedback, detailed logging, and graceful failure handling for your blog API.

🎯 What We're Building

Your error handling system will provide:

• 🎯 Custom exception hierarchy with detailed context
• 📊 Structured error responses with helpful information
• 📝 Comprehensive logging for debugging and monitoring
• 🔍 Request tracking with unique identifiers
• 🛡️ Security event logging for audit trails

🏗️ Error Handling Architecture

1. Exception Hierarchy

BlogAPIException (Base)
├── ValidationException
├── AuthenticationException  
├── AuthorizationException
├── NotFoundException
├── ConflictException
├── RateLimitException
└── ServerErrorException

2. Error Response Format

{
  "error": {
    "type": "validation",
    "code": "invalid_format", 
    "message": "Email format is invalid",
    "details": {
      "field": "email",
      "provided_value": "invalid-email"
    },
    "timestamp": "2024-01-15T10:30:00Z",
    "request_id": "req_123456789",
    "help": "Please provide a valid email address"
  }
}

3. Error Categories

Client Errors (4xx)

• Validation: Malformed requests, invalid data
• Authentication: Missing/invalid credentials
• Authorization: Insufficient permissions
• Not Found: Missing resources
• Conflict: Duplicate resources, constraint violations

Server Errors (5xx)

• Database: Connection failures, constraint violations
• External Services: Third-party API failures
• Internal: Unexpected application errors

🎨 Custom Exception Design

Base Exception Class

class BlogAPIException(Exception):
    def __init__(self, message: str, error_type: ErrorType, 
                 error_code: ErrorCode, status_code: int = 400,
                 details: Optional[Dict] = None):
        super().__init__(message)
        self.message = message
        self.error_type = error_type
        self.error_code = error_code  
        self.status_code = status_code
        self.details = details or {}
        self.timestamp = datetime.utcnow().isoformat()
        self.request_id = str(uuid.uuid4())
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            'error': {
                'type': self.error_type.value,
                'code': self.error_code.value,
                'message': self.message,
                'details': self.details,
                'timestamp': self.timestamp,
                'request_id': self.request_id
            }
        }

Specific Exception Classes

class ValidationException(BlogAPIException):
    def __init__(self, message: str, field: str = None, 
                 value: Any = None, details: Optional[Dict] = None):
        if not details:
            details = {}
        
        if field:
            details['field'] = field
        if value is not None:
            details['provided_value'] = str(value)
        
        super().__init__(
            message=message,
            error_type=ErrorType.VALIDATION,
            error_code=ErrorCode.INVALID_FORMAT,
            status_code=400,
            details=details
        )

class NotFoundException(BlogAPIException):
    def __init__(self, resource: str, identifier: Any = None):
        details = {'resource_type': resource}
        if identifier is not None:
            details['identifier'] = str(identifier)
        
        message = f"{resource.title()} not found"
        if identifier:
            message += f" (ID: {identifier})"
        
        super().__init__(
            message=message,
            error_type=ErrorType.NOT_FOUND,
            error_code=self._get_error_code(resource),
            status_code=404,
            details=details
        )

📝 Professional Logging

Structured Logging Setup

class LoggingManager:
    def __init__(self, log_level: str = 'INFO'):
        self.setup_loggers()
    
    def setup_loggers(self):
        # Main application logger
        self.logger = logging.getLogger('blog_api')
        
        # Specialized loggers
        self.error_logger = logging.getLogger('blog_api.errors')
        self.security_logger = logging.getLogger('blog_api.security')
        self.request_logger = logging.getLogger('blog_api.requests')
        
        # Configure formatters
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        
        # Add handlers (console, file, external services)
        self.setup_handlers(formatter)

Error Context Logging

def log_error(self, error: Exception, context: Optional[Dict] = None):
    """Log error with full context"""
    error_data = {
        'error_type': type(error).__name__,
        'error_message': str(error),
        'traceback': traceback.format_exc(),
        'context': context or {}
    }
    
    if isinstance(error, BlogAPIException):
        error_data.update({
            'error_code': error.error_code.value,
            'status_code': error.status_code,
            'request_id': error.request_id,
            'error_details': error.details
        })
    
    self.error_logger.error(f"Error occurred: {str(error)}", extra=error_data)

Security Event Logging

def log_security_event(self, event_type: str, details: Dict):
    """Log security-related events"""
    security_data = {
        'event_type': event_type,
        'timestamp': datetime.utcnow().isoformat(),
        'severity': self._get_severity(event_type),
        **details
    }
    
    self.security_logger.warning(
        f"Security Event: {event_type}", 
        extra=security_data
    )

# Usage examples
logging_manager.log_security_event('failed_login', {
    'username': 'admin',
    'ip_address': '192.168.1.100',
    'user_agent': 'Mozilla/5.0...',
    'attempt_count': 3
})

logging_manager.log_security_event('permission_denied', {
    'user_id': 123,
    'required_permission': 'admin',
    'resource': '/admin/users',
    'action': 'DELETE'
})

🔄 Error Response Formatting

Response Formatter

class ErrorResponseFormatter:
    def format_error(self, error: BlogAPIException, 
                    request_id: str = None) -> Dict[str, Any]:
        """Format error for API response"""
        response = error.to_dict()
        
        # Override request ID if provided
        if request_id:
            response['error']['request_id'] = request_id
        
        # Add contextual help messages
        help_messages = {
            ErrorType.VALIDATION: "Check the provided data and try again",
            ErrorType.AUTHENTICATION: "Please log in and try again", 
            ErrorType.AUTHORIZATION: "You don't have permission for this action",
            ErrorType.NOT_FOUND: "The requested resource was not found",
            ErrorType.RATE_LIMIT: "Please wait before making more requests"
        }
        
        if error.error_type in help_messages:
            response['error']['help'] = help_messages[error.error_type]
        
        # Add support links for complex errors
        if error.error_type == ErrorType.VALIDATION:
            response['error']['docs'] = "https://docs.example.com/validation"
        
        return response
    
    def format_validation_errors(self, errors: List[Dict]) -> Dict[str, Any]:
        """Format multiple validation errors"""
        return {
            'error': {
                'type': ErrorType.VALIDATION.value,
                'code': ErrorCode.INVALID_FORMAT.value,
                'message': f'{len(errors)} validation errors occurred',
                'details': {
                    'validation_errors': errors,
                    'error_count': len(errors)
                },
                'timestamp': datetime.utcnow().isoformat(),
                'help': 'Fix all validation errors and try again'
            }
        }

Development vs Production Responses

def format_error(self, error: BlogAPIException, debug: bool = False):
    """Format with different detail levels"""
    response = error.to_dict()
    
    if debug:
        # Development: Include stack traces and internal details
        response['error']['stack_trace'] = traceback.format_exc()
        response['error']['internal_details'] = error.details
    else:
        # Production: Hide sensitive information
        if error.error_type == ErrorType.SERVER_ERROR:
            response['error']['message'] = "An internal error occurred"
            response['error']['details'] = {}
    
    return response

🛠️ FastAPI Integration

Global Exception Handler

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(BlogAPIException)
async def blog_api_exception_handler(request: Request, exc: BlogAPIException):
    """Handle custom blog API exceptions"""
    context = {
        'method': request.method,
        'url': str(request.url),
        'headers': dict(request.headers),
        'user_agent': request.headers.get('user-agent')
    }
    
    response = error_handler.handle_exception(exc, context)
    return JSONResponse(
        status_code=exc.status_code,
        content=response
    )

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    """Handle unexpected exceptions"""
    context = {
        'method': request.method,
        'url': str(request.url),
        'exception_type': type(exc).__name__
    }
    
    response = error_handler.handle_exception(exc, context)
    return JSONResponse(
        status_code=500,
        content=response
    )

Request Middleware with Error Tracking

import time
from starlette.middleware.base import BaseHTTPMiddleware

class ErrorTrackingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        request_id = str(uuid.uuid4())
        
        # Add request ID to request state
        request.state.request_id = request_id
        
        try:
            response = await call_next(request)
            
            # Log successful requests
            duration = time.time() - start_time
            logging_manager.log_request(
                method=request.method,
                path=request.url.path,
                status_code=response.status_code,
                duration=duration,
                request_id=request_id
            )
            
            return response
            
        except Exception as exc:
            # Log failed requests
            duration = time.time() - start_time
            logging_manager.log_error(exc, {
                'request_id': request_id,
                'method': request.method,
                'path': request.url.path,
                'duration': duration
            })
            
            raise exc

# Add middleware to app
app.add_middleware(ErrorTrackingMiddleware)

🎯 Helper Functions

Convenient Error Raising

def raise_validation_error(message: str, field: str = None, value: Any = None):
    """Raise validation error with context"""
    raise ValidationException(message, field=field, value=value)

def raise_not_found(resource: str, identifier: Any = None):
    """Raise not found error"""
    raise NotFoundException(resource, identifier)

def raise_permission_denied(permission: str = None, resource: str = None):
    """Raise authorization error with context"""
    message = "You don't have permission to perform this action"
    details = {}
    if permission:
        details['required_permission'] = permission
    if resource:
        details['resource'] = resource
    
    raise AuthorizationException(message, required_permission=permission)

# Usage in endpoints
@app.get("/posts/{post_id}")
async def get_post(post_id: int):
    post = await db.get_post(post_id)
    if not post:
        raise_not_found("post", post_id)
    return post

@app.delete("/posts/{post_id}")
async def delete_post(post_id: int, current_user: dict = Depends(get_current_user)):
    post = await db.get_post(post_id)
    if not post:
        raise_not_found("post", post_id)
    
    if post.author_id != current_user['id'] and current_user['role'] != 'admin':
        raise_permission_denied("delete_post", f"post/{post_id}")
    
    await db.delete_post(post_id)
    return {"message": "Post deleted"}

📊 Error Monitoring and Alerting

Error Rate Tracking

class ErrorMetrics:
    def __init__(self):
        self.error_counts = defaultdict(int)
        self.error_rates = {}
        self.last_reset = time.time()
    
    def record_error(self, error_type: str, error_code: str):
        """Record error occurrence"""
        key = f"{error_type}:{error_code}"
        self.error_counts[key] += 1
        
        # Calculate error rates
        self.calculate_rates()
    
    def calculate_rates(self):
        """Calculate error rates per minute"""
        current_time = time.time()
        time_window = current_time - self.last_reset
        
        if time_window >= 60:  # Reset every minute
            for key, count in self.error_counts.items():
                self.error_rates[key] = count / (time_window / 60)
            
            self.error_counts.clear()
            self.last_reset = current_time
    
    def get_high_error_rates(self, threshold: float = 10.0) -> List[Dict]:
        """Get error types exceeding threshold"""
        high_errors = []
        for key, rate in self.error_rates.items():
            if rate > threshold:
                error_type, error_code = key.split(':', 1)
                high_errors.append({
                    'error_type': error_type,
                    'error_code': error_code,
                    'rate_per_minute': rate
                })
        return high_errors

Alerting Integration

class AlertManager:
    def __init__(self, webhook_url: str = None):
        self.webhook_url = webhook_url
    
    async def send_error_alert(self, error: BlogAPIException, context: Dict):
        """Send alert for critical errors"""
        if self.should_alert(error):
            alert_data = {
                'severity': 'high' if error.status_code >= 500 else 'medium',
                'error_type': error.error_type.value,
                'error_code': error.error_code.value,
                'message': error.message,
                'request_id': error.request_id,
                'context': context
            }
            
            await self.send_webhook_alert(alert_data)
    
    def should_alert(self, error: BlogAPIException) -> bool:
        """Determine if error should trigger alert"""
        # Alert on server errors
        if error.status_code >= 500:
            return True
        
        # Alert on security events
        if error.error_type in [ErrorType.AUTHORIZATION, ErrorType.AUTHENTICATION]:
            return True
        
        # Alert on high error rates
        if self.is_high_frequency_error(error):
            return True
        
        return False

💡 Best Practices

1. Error Message Guidelines

# ✅ Good: Clear and actionable
"Email format is invalid. Please provide a valid email address."

# ❌ Bad: Vague and unhelpful  
"Invalid input"

# ✅ Good: Specific field information
"Username must be 3-20 characters and contain only letters, numbers, and underscores"

# ❌ Bad: No context
"Validation failed"

2. Security Considerations

# ✅ Safe: Don't reveal system internals
"User not found"

# ❌ Dangerous: Reveals database structure
"SELECT * FROM users WHERE id = 123 returned no rows"

# ✅ Safe: Generic server error message
"An internal error occurred. Please try again later."

# ❌ Dangerous: Exposes implementation details
"Redis connection failed: Connection refused on localhost:6379"

3. Error Recovery Guidance

def format_error_with_recovery(self, error: BlogAPIException):
    response = error.to_dict()
    
    # Add recovery suggestions
    recovery_actions = {
        ErrorCode.TOKEN_EXPIRED: ["Refresh your authentication token", "Log in again"],
        ErrorCode.RATE_LIMIT_EXCEEDED: ["Wait before retrying", "Reduce request frequency"],
        ErrorCode.VALIDATION_FAILED: ["Check required fields", "Verify data formats"]
    }
    
    if error.error_code in recovery_actions:
        response['error']['recovery_actions'] = recovery_actions[error.error_code]
    
    return response

🔍 Testing Error Handling

Unit Tests for Exceptions

def test_validation_exception():
    with pytest.raises(ValidationException) as exc_info:
        raise_validation_error("Invalid email", field="email", value="invalid")
    
    error = exc_info.value
    assert error.error_type == ErrorType.VALIDATION
    assert error.details['field'] == 'email'
    assert error.details['provided_value'] == 'invalid'

def test_error_formatting():
    error = NotFoundException("user", 123)
    formatted = formatter.format_error(error)
    
    assert formatted['error']['type'] == 'not_found'
    assert formatted['error']['code'] == 'user_not_found'
    assert '123' in formatted['error']['message']

Integration Tests

async def test_error_endpoint_responses():
    # Test validation error response
    response = await client.post('/users', json={'invalid': 'data'})
    assert response.status_code == 400
    assert response.json()['error']['type'] == 'validation'
    
    # Test not found error response
    response = await client.get('/users/999999')
    assert response.status_code == 404
    assert 'not found' in response.json()['error']['message']

🎯 Your Implementation Tasks

1. Create exception hierarchy with base BlogAPIException
2. Build specific exception classes for different error types
3. Implement error formatters with helpful responses
4. Set up comprehensive logging with multiple loggers
5. Create helper functions for common error scenarios
6. Test thoroughly with various error conditions

🚀 Production Considerations

• Log aggregation (ELK stack, Splunk, CloudWatch)
• Error monitoring (Sentry, Rollbar, Bugsnag)
• Alerting systems (PagerDuty, Slack webhooks)
• Rate limiting to prevent error spam
• Error budgets and SLA monitoring
• User-friendly error pages for web interfaces

Ready to build bulletproof error handling? Let's create a system that fails gracefully! ❗