# apps/core/mixins/error_format_mixin.py from rest_framework import serializers, exceptions from rest_framework.response import Response from rest_framework import status from apps.core.utils import tz_now class ErrorFormatMixin: """ Dynamic mixin for standardized error formatting across all ViewSets """ # Error codes mapping ERROR_CODES = { 'null': '1000', 'blank': '1002', 'required': '1001', 'invalid': '1003', 'unique': '1004', 'max_length': '1005', 'min_length': '1006', 'max_value': '1007', 'min_value': '1008', 'does_not_exist': '1009', 'incorrect_type': '1010' } @property def _tz_now_formatted(self): """Get current time in timezone aware format using utils function""" return tz_now() def get_error_code(self, error_message): """ Extract specific error code from error message dynamically """ error_message_lower = str(error_message).lower() if any(phrase in error_message_lower for phrase in ['null', 'may not be null']): return self.ERROR_CODES['null'] elif 'blank' in error_message_lower: return self.ERROR_CODES['blank'] elif 'required' in error_message_lower: return self.ERROR_CODES['required'] elif 'unique' in error_message_lower: return self.ERROR_CODES['unique'] elif 'maximum length' in error_message_lower or 'max_length' in error_message_lower: return self.ERROR_CODES['max_length'] elif 'minimum length' in error_message_lower or 'min_length' in error_message_lower: return self.ERROR_CODES['min_length'] elif 'invalid' in error_message_lower: return self.ERROR_CODES['invalid'] elif 'does not exist' in error_message_lower: return self.ERROR_CODES['does_not_exist'] elif 'incorrect type' in error_message_lower: return self.ERROR_CODES['incorrect_type'] else: return "1000" # Default error code def _generate_error_message(self, error_details): """ Generate a proper summary error message from error details """ if not error_details: return "Please fix the validation errors" # Count errors by field field_errors = {} for error in error_details: field = error.get('field', 'non_field_errors') if field not in field_errors: field_errors[field] = [] field_errors[field].append(error.get('description', '')) # Generate summary message if len(field_errors) == 1: field_name = list(field_errors.keys())[0] if field_name == 'non_field_errors': return field_errors[field_name][0] # Use the first non-field error else: return f"Please check the {field_name} field" elif len(field_errors) <= 3: field_names = list(field_errors.keys()) # Remove 'non_field_errors' from field names for display display_fields = [f for f in field_names if f != 'non_field_errors'] if display_fields: if len(display_fields) == 1: return f"Please check the {display_fields[0]} field" else: fields_str = ", ".join(display_fields[:-1]) + f" and {display_fields[-1]}" return f"Please check the {fields_str} fields" else: # Only non-field errors return field_errors['non_field_errors'][0] else: return f"Please check {len(field_errors)} fields with errors" def format_validation_errors(self, errors, message=None, code="1000"): """ Format DRF validation errors into standardized format """ error_details = [] def process_errors(field_path, error_data): """ Recursively process nested validation errors """ if isinstance(error_data, dict): # Nested serializer errors for nested_field, nested_errors in error_data.items(): nested_field_path = f"{field_path}.{nested_field}" if field_path else nested_field process_errors(nested_field_path, nested_errors) elif isinstance(error_data, list): # List of errors for a field for error in error_data: if isinstance(error, dict): # Nested errors in list process_errors(field_path, error) else: # Simple error message error_code = self.get_error_code(error) error_details.append({ "field": field_path or "non_field_errors", "issue": "validation_error", "description": str(error), "code": error_code }) else: # Single error message error_code = self.get_error_code(error_data) error_details.append({ "field": field_path or "non_field_errors", "issue": "validation_error", "description": str(error_data), "code": error_code }) # Start processing from root process_errors("", errors) # If no specific field errors, check for non_field_errors if not error_details and 'non_field_errors' in errors: process_errors("non_field_errors", errors['non_field_errors']) # Generate dynamic message if not provided if message is None: message = self._generate_error_message(error_details) return { "success": False, "error": { "code": code, "details": error_details }, "message": message, "timestamp": self._tz_now_formatted } def _get_first_error_description(self, error_details): """ Extract the first error description from error details """ if error_details and isinstance(error_details, list) and len(error_details) > 0: return error_details[0].get('description', 'An error occurred') return 'An error occurred' def handle_exception(self, exc): """ Override DRF's exception handler to use standardized error format """ # Handle validation errors if isinstance(exc, (exceptions.ValidationError, serializers.ValidationError)): # Format the errors to get the structured details error_data = self.format_validation_errors( exc.detail if hasattr(exc, 'detail') else str(exc), message=None # Let format_validation_errors generate the message dynamically ) return Response(error_data, status=status.HTTP_400_BAD_REQUEST) # Handle other DRF exceptions if isinstance(exc, exceptions.APIException): # Extract the actual error message from exc.detail error_message = str(exc.detail) if hasattr(exc, 'detail') else "API Error occurred" error_data = { "success": False, "error": { "code": "4000", "details": [{ "field": "non_field_errors", "issue": "api_error", "description": error_message, "code": str(exc.status_code) }] }, "message": error_message, "timestamp": self._tz_now_formatted } return Response(error_data, status=exc.status_code) # Handle generic exceptions error_message = str(exc) error_data = { "success": False, "error": { "code": "5000", "details": [{ "field": "non_field_errors", "issue": "server_error", "description": error_message, "code": "5000" }] }, "message": error_message, "timestamp": self._tz_now_formatted } return Response(error_data, status=status.HTTP_500_INTERNAL_SERVER_ERROR)