""" Comprehensive error handling for BaoLife. Provides: - Custom exception classes with error codes and retry logic - Error decorator for automatic error handling - Client error messaging integration - Structured logging for all errors """ import logging import traceback import asyncio from functools import wraps from typing import Dict, Any, Optional, Callable from datetime import datetime logger = logging.getLogger(__name__) class GameError(Exception): """ Base exception class for all game-related errors. Attributes: message: Human-readable error message error_code: Machine-readable error code retry_possible: Whether the operation can be retried """ def __init__(self, message: str, error_code: str = 'GAME_ERROR', retry_possible: bool = False): self.message = message self.error_code = error_code self.retry_possible = retry_possible super().__init__(self.message) class InsufficientResourcesError(GameError): """ Raised when player lacks required resources (energy, money, etc). Attributes: resource_type: Type of resource (energy, money, diamonds, etc) required: Amount required current: Amount player currently has """ def __init__(self, resource_type: str, required: float, current: float): self.resource_type = resource_type self.required = required self.current = current message = f"Insufficient {resource_type}: need {required}, have {current}" super().__init__(message, error_code='INSUFFICIENT_RESOURCES', retry_possible=True) class ServerError(GameError): """ Raised for internal server errors that should be logged and monitored. Used for database failures, external API failures, etc. """ def __init__(self, message: str, original_exception: Optional[Exception] = None): self.original_exception = original_exception super().__init__(message, error_code='SERVER_ERROR', retry_possible=True) def handle_errors(send_to_client: Optional[Callable] = None, log_level: str = 'ERROR'): """ Decorator for automatic error handling with logging and client notification. Args: send_to_client: Optional function to send errors to client log_level: Logging level for errors (ERROR, WARNING, INFO) Usage: @handle_errors(send_to_client=send_error_func) def some_function(player_id, ...): ... @handle_errors() # Just logging, no client notification async def async_function(player_id, ...): ... """ def decorator(func): @wraps(func) async def async_wrapper(*args, **kwargs): player_id = kwargs.get('player_id') or (args[0] if args else None) try: return await func(*args, **kwargs) except GameError as e: # Log game errors at specified level log_func = getattr(logger, log_level.lower(), logger.error) log_func( f"GameError in {func.__name__}: {e.message}", extra={ 'user_id': player_id, 'error_code': e.error_code, 'retry_possible': e.retry_possible, 'function': func.__name__ } ) # Send to client if function provided if send_to_client and player_id: await send_error_to_client( player_id, e.message, e.error_code, e.retry_possible, send_to_client ) # Re-raise to allow calling code to handle if needed raise except Exception as e: # Log unexpected errors with full traceback logger.error( f"Unexpected error in {func.__name__}: {str(e)}", exc_info=True, extra={ 'user_id': player_id, 'function': func.__name__, 'traceback': traceback.format_exc() } ) # Send generic error to client if send_to_client and player_id: await send_error_to_client( player_id, "An unexpected error occurred. Please try again.", 'UNEXPECTED_ERROR', True, send_to_client ) # Re-raise as ServerError raise ServerError("Unexpected error occurred", original_exception=e) @wraps(func) def sync_wrapper(*args, **kwargs): player_id = kwargs.get('player_id') or (args[0] if args else None) try: return func(*args, **kwargs) except GameError as e: # Log game errors at specified level log_func = getattr(logger, log_level.lower(), logger.error) log_func( f"GameError in {func.__name__}: {e.message}", extra={ 'user_id': player_id, 'error_code': e.error_code, 'retry_possible': e.retry_possible, 'function': func.__name__ } ) # Note: Sync wrapper cannot await send_to_client # Client notification must be handled by caller for sync functions # Re-raise to allow calling code to handle if needed raise except Exception as e: # Log unexpected errors with full traceback logger.error( f"Unexpected error in {func.__name__}: {str(e)}", exc_info=True, extra={ 'user_id': player_id, 'function': func.__name__, 'traceback': traceback.format_exc() } ) # Re-raise as ServerError raise ServerError("Unexpected error occurred", original_exception=e) # Return appropriate wrapper based on whether function is async if asyncio.iscoroutinefunction(func): return async_wrapper else: return sync_wrapper return decorator async def send_error_to_client( player_id: Any, message: str, error_code: str, retry_possible: bool, send_func: Callable ): """ Send formatted error message to client via WebSocket. Args: player_id: Player identifier message: Human-readable error message error_code: Machine-readable error code retry_possible: Whether operation can be retried send_func: Async function to send message to client """ error_payload = { 'type': 'error', 'error_code': error_code, 'message': message, 'retry_possible': retry_possible, 'timestamp': datetime.utcnow().isoformat() } try: await send_func(player_id, error_payload) except Exception as e: logger.error( f"Failed to send error to client {player_id}: {e}", extra={'user_id': player_id} ) # Example usage functions async def example_purchase_energy(player_id: int, amount: int, send_to_client: Callable) -> Dict[str, Any]: """ Example function showing error handling usage. This demonstrates: - InsufficientResourcesError for validation - Automatic error logging and client notification via decorator - Returning success response """ @handle_errors(send_to_client=send_to_client) async def _purchase(player_id: int, amount: int) -> Dict[str, Any]: # Simulated player state player_money = 50 cost = amount * 10 if player_money < cost: raise InsufficientResourcesError('money', cost, player_money) # Process purchase... return { 'success': True, 'message': f'Purchased {amount} energy', 'new_balance': player_money - cost } return await _purchase(player_id, amount) def example_validate_action(player_id: int, action: str) -> bool: """ Example synchronous function showing error handling. Demonstrates error decorator on sync functions. """ @handle_errors(log_level='WARNING') def _validate(player_id: int, action: str) -> bool: if action not in ['eat', 'sleep', 'work', 'socialize']: raise GameError( f"Invalid action: {action}", error_code='INVALID_ACTION', retry_possible=False ) return True return _validate(player_id, action)