"""Centralized error handling service for consistent API responses.""" from typing import Any from flask import jsonify class ErrorHandlingService: """Service for standardized error handling and responses.""" @staticmethod def handle_validation_error(error: ValueError) -> tuple[Any, int]: """Handle validation errors consistently.""" error_str = str(error) # Map common validation errors to appropriate HTTP status codes status_code = 400 if "not found" in error_str.lower(): status_code = 404 elif ( "not authorized" in error_str.lower() or "permission" in error_str.lower() ): status_code = 403 elif ( "already exists" in error_str.lower() or "already linked" in error_str.lower() ): status_code = 409 elif ( "not configured" in error_str.lower() or "cannot unlink" in error_str.lower() ): status_code = 400 elif "not deletable" in error_str.lower(): status_code = 403 return jsonify({"error": error_str}), status_code @staticmethod def handle_generic_error(error: Exception) -> tuple[Any, int]: """Handle generic exceptions with 500 status.""" return jsonify({"error": str(error)}), 500 @staticmethod def handle_service_result(result: dict) -> tuple[Any, int]: """Handle service method results that return success/error dictionaries.""" if result.get("success"): return jsonify(result), 200 return jsonify(result), 400 @staticmethod def create_success_response( message: str, data: dict = None, status_code: int = 200, ) -> tuple[Any, int]: """Create a standardized success response.""" response = {"message": message} if data: response.update(data) return jsonify(response), status_code @staticmethod def create_error_response( message: str, status_code: int = 400, details: dict = None, ) -> tuple[Any, int]: """Create a standardized error response.""" response = {"error": message} if details: response.update(details) return jsonify(response), status_code @staticmethod def handle_auth_error(error_type: str) -> tuple[Any, int]: """Handle common authentication errors.""" auth_errors = { "user_not_authenticated": ("User not authenticated", 401), "user_not_found": ("User not found", 404), "invalid_credentials": ("Invalid credentials", 401), "account_disabled": ("Account is disabled", 401), "insufficient_credits": ("Insufficient credits", 402), "admin_required": ("Admin privileges required", 403), } if error_type in auth_errors: message, status = auth_errors[error_type] return jsonify({"error": message}), status return jsonify({"error": "Authentication error"}), 401 @staticmethod def handle_file_operation_error( operation: str, error: Exception ) -> tuple[Any, int]: """Handle file operation errors consistently.""" error_message = f"Failed to {operation}: {error!s}" # Check for specific file operation errors if ( "not found" in str(error).lower() or "no such file" in str(error).lower() ): return jsonify({"error": f"File not found during {operation}"}), 404 if "permission" in str(error).lower(): return jsonify( {"error": f"Permission denied during {operation}"} ), 403 return jsonify({"error": error_message}), 500 @staticmethod def wrap_service_call(service_func, *args, **kwargs) -> tuple[Any, int]: """Wrap service calls with standardized error handling.""" try: result = service_func(*args, **kwargs) # If result is a dictionary with success/error structure if isinstance(result, dict) and "success" in result: return ErrorHandlingService.handle_service_result(result) # If result is a simple dictionary (like user data) if isinstance(result, dict): return jsonify(result), 200 # For other types, assume success return jsonify({"result": result}), 200 except ValueError as e: return ErrorHandlingService.handle_validation_error(e) except Exception as e: return ErrorHandlingService.handle_generic_error(e)