""" API Response - Standardized API response structure. Provides consistent success/error response formats for all API endpoints. Clients can use a unified handler for both success and error responses. Success Response: { "success": true, "message": "Operation completed successfully", "data": { ... } } Error Response: { "success": false, "error": { "code": "INSUFFICIENT_CREDITS", "message": "You don't have enough credits", "details": { ... } } } Usage: # In routers - raising errors from core.api_response import APIError, ErrorCode raise APIError( code=ErrorCode.INSUFFICIENT_CREDITS, message="You need at least 10 credits", status_code=402, details={"required": 10, "available": 5} ) # In routers - success response from core.api_response import success_response return success_response( data={"job_id": "123", "status": "queued"}, message="Job created successfully" ) """ from typing import Optional, Any, Dict from pydantic import BaseModel, Field # ============================================================================= # Error Codes - Machine-readable error identifiers # ============================================================================= class ErrorCode: """Standard error codes for consistent client handling.""" # Authentication errors (401) UNAUTHORIZED = "UNAUTHORIZED" TOKEN_EXPIRED = "TOKEN_EXPIRED" TOKEN_INVALID = "TOKEN_INVALID" # Authorization errors (403) FORBIDDEN = "FORBIDDEN" ADMIN_REQUIRED = "ADMIN_REQUIRED" # Payment/Credits errors (402) INSUFFICIENT_CREDITS = "INSUFFICIENT_CREDITS" PAYMENT_REQUIRED = "PAYMENT_REQUIRED" PAYMENT_FAILED = "PAYMENT_FAILED" # Resource errors (404) NOT_FOUND = "NOT_FOUND" USER_NOT_FOUND = "USER_NOT_FOUND" JOB_NOT_FOUND = "JOB_NOT_FOUND" # Validation errors (400, 422) VALIDATION_ERROR = "VALIDATION_ERROR" BAD_REQUEST = "BAD_REQUEST" INVALID_INPUT = "INVALID_INPUT" # Rate limiting (429) RATE_LIMITED = "RATE_LIMITED" TOO_MANY_REQUESTS = "TOO_MANY_REQUESTS" # Service errors (5xx) SERVER_ERROR = "SERVER_ERROR" SERVICE_UNAVAILABLE = "SERVICE_UNAVAILABLE" EXTERNAL_SERVICE_ERROR = "EXTERNAL_SERVICE_ERROR" # Business logic errors JOB_ALREADY_PROCESSING = "JOB_ALREADY_PROCESSING" JOB_EXPIRED = "JOB_EXPIRED" OPERATION_NOT_ALLOWED = "OPERATION_NOT_ALLOWED" # ============================================================================= # Pydantic Models - For OpenAPI documentation # ============================================================================= class ErrorDetail(BaseModel): """Structured error information.""" code: str = Field(..., description="Machine-readable error code") message: str = Field(..., description="Human-readable error message") details: Optional[Dict[str, Any]] = Field( None, description="Additional context about the error" ) class ApiErrorResponse(BaseModel): """Standard error response format.""" success: bool = Field(False, description="Always false for errors") error: ErrorDetail = Field(..., description="Error details") class ApiSuccessResponse(BaseModel): """Standard success response format.""" success: bool = Field(True, description="Always true for success") message: Optional[str] = Field(None, description="Optional success message") data: Optional[Dict[str, Any]] = Field(None, description="Response payload") # ============================================================================= # Custom Exception - For raising API errors # ============================================================================= class APIError(Exception): """ Custom exception for API errors with structured response. Raise this exception anywhere in your code, and the global exception handler will convert it to a standardized JSON response. Example: raise APIError( code=ErrorCode.INSUFFICIENT_CREDITS, message="You need at least 10 credits for this operation", status_code=402, details={"required": 10, "available": 5} ) """ def __init__( self, code: str, message: str, status_code: int = 400, details: Optional[Dict[str, Any]] = None ): self.code = code self.message = message self.status_code = status_code self.details = details super().__init__(message) def to_dict(self) -> dict: """Convert to response dictionary.""" return error_response(self.code, self.message, self.details) # ============================================================================= # Helper Functions - For building responses # ============================================================================= def success_response( data: Optional[Dict[str, Any]] = None, message: Optional[str] = None ) -> dict: """ Create a standardized success response. Args: data: Response payload (dict) message: Optional success message Returns: dict: {"success": true, "message": "...", "data": {...}} Example: return success_response( data={"job_id": "123", "status": "queued"}, message="Job created successfully" ) """ response = {"success": True} if message: response["message"] = message if data is not None: response["data"] = data return response def error_response( code: str, message: str, details: Optional[Dict[str, Any]] = None ) -> dict: """ Create a standardized error response. Args: code: Machine-readable error code (use ErrorCode constants) message: Human-readable error message details: Optional additional context Returns: dict: {"success": false, "error": {"code": "...", "message": "...", "details": {...}}} Example: return error_response( code=ErrorCode.NOT_FOUND, message="Job not found", details={"job_id": "xyz"} ) """ error = { "code": code, "message": message } if details is not None: error["details"] = details return { "success": False, "error": error } def status_to_error_code(status_code: int) -> str: """ Map HTTP status code to a default error code. Args: status_code: HTTP status code Returns: str: Corresponding error code """ mapping = { 400: ErrorCode.BAD_REQUEST, 401: ErrorCode.UNAUTHORIZED, 402: ErrorCode.PAYMENT_REQUIRED, 403: ErrorCode.FORBIDDEN, 404: ErrorCode.NOT_FOUND, 422: ErrorCode.VALIDATION_ERROR, 429: ErrorCode.RATE_LIMITED, 500: ErrorCode.SERVER_ERROR, 502: ErrorCode.EXTERNAL_SERVICE_ERROR, 503: ErrorCode.SERVICE_UNAVAILABLE, } return mapping.get(status_code, ErrorCode.SERVER_ERROR) __all__ = [ # Error codes 'ErrorCode', # Pydantic models 'ErrorDetail', 'ApiErrorResponse', 'ApiSuccessResponse', # Exception 'APIError', # Helper functions 'success_response', 'error_response', 'status_to_error_code', ]