346 lines
10 KiB
Python
346 lines
10 KiB
Python
"""
|
|
Comprehensive error response schemas for consistent API responses.
|
|
"""
|
|
from typing import Any, Dict, List, Optional, Union
|
|
from pydantic import BaseModel, Field
|
|
from enum import Enum
|
|
from datetime import datetime
|
|
|
|
|
|
class ErrorCode(str, Enum):
|
|
"""Standard error codes for the application."""
|
|
|
|
# Validation errors
|
|
VALIDATION_ERROR = "validation_error"
|
|
FIELD_REQUIRED = "field_required"
|
|
FIELD_INVALID = "field_invalid"
|
|
|
|
# Authentication & Authorization
|
|
UNAUTHORIZED = "unauthorized"
|
|
FORBIDDEN = "forbidden"
|
|
TOKEN_EXPIRED = "token_expired"
|
|
TOKEN_INVALID = "token_invalid"
|
|
|
|
# Resource errors
|
|
NOT_FOUND = "not_found"
|
|
ALREADY_EXISTS = "already_exists"
|
|
CONFLICT = "conflict"
|
|
|
|
# Database errors
|
|
DATABASE_ERROR = "database_error"
|
|
INTEGRITY_ERROR = "integrity_error"
|
|
TRANSACTION_ERROR = "transaction_error"
|
|
|
|
# Business logic errors
|
|
BUSINESS_RULE_VIOLATION = "business_rule_violation"
|
|
INVALID_OPERATION = "invalid_operation"
|
|
|
|
# System errors
|
|
INTERNAL_ERROR = "internal_error"
|
|
SERVICE_UNAVAILABLE = "service_unavailable"
|
|
TIMEOUT = "timeout"
|
|
|
|
# File & Import errors
|
|
FILE_TOO_LARGE = "file_too_large"
|
|
INVALID_FILE_FORMAT = "invalid_file_format"
|
|
IMPORT_ERROR = "import_error"
|
|
|
|
# Security errors
|
|
SECURITY_VIOLATION = "security_violation"
|
|
RATE_LIMITED = "rate_limited"
|
|
|
|
|
|
class ErrorDetail(BaseModel):
|
|
"""Individual error detail with context."""
|
|
|
|
code: ErrorCode = Field(..., description="Specific error code")
|
|
message: str = Field(..., description="Human-readable error message")
|
|
field: Optional[str] = Field(None, description="Field name if field-specific error")
|
|
context: Optional[Dict[str, Any]] = Field(None, description="Additional error context")
|
|
|
|
|
|
class ValidationErrorDetail(ErrorDetail):
|
|
"""Specialized error detail for validation errors."""
|
|
|
|
field: str = Field(..., description="Field that failed validation")
|
|
input_value: Optional[Any] = Field(None, description="Value that failed validation")
|
|
constraint: Optional[str] = Field(None, description="Validation constraint that failed")
|
|
|
|
|
|
class ErrorResponse(BaseModel):
|
|
"""Standard error response schema."""
|
|
|
|
success: bool = Field(False, description="Always false for error responses")
|
|
error: ErrorDetail = Field(..., description="Primary error information")
|
|
errors: Optional[List[ErrorDetail]] = Field(None, description="Additional errors if multiple")
|
|
timestamp: datetime = Field(default_factory=datetime.utcnow, description="Error timestamp")
|
|
request_id: Optional[str] = Field(None, description="Request identifier for tracing")
|
|
|
|
class Config:
|
|
json_encoders = {
|
|
datetime: lambda v: v.isoformat()
|
|
}
|
|
|
|
|
|
class ValidationErrorResponse(ErrorResponse):
|
|
"""Specialized error response for validation failures."""
|
|
|
|
error: ValidationErrorDetail = Field(..., description="Primary validation error")
|
|
errors: Optional[List[ValidationErrorDetail]] = Field(None, description="Additional validation errors")
|
|
|
|
|
|
class SuccessResponse(BaseModel):
|
|
"""Standard success response schema."""
|
|
|
|
success: bool = Field(True, description="Always true for success responses")
|
|
data: Optional[Any] = Field(None, description="Response data")
|
|
message: Optional[str] = Field(None, description="Optional success message")
|
|
timestamp: datetime = Field(default_factory=datetime.utcnow, description="Response timestamp")
|
|
|
|
class Config:
|
|
json_encoders = {
|
|
datetime: lambda v: v.isoformat()
|
|
}
|
|
|
|
|
|
class PaginatedResponse(SuccessResponse):
|
|
"""Standard paginated response schema."""
|
|
|
|
data: List[Any] = Field(..., description="Paginated data items")
|
|
pagination: Dict[str, Any] = Field(..., description="Pagination metadata")
|
|
|
|
@classmethod
|
|
def create(
|
|
cls,
|
|
items: List[Any],
|
|
total: int,
|
|
page: int,
|
|
page_size: int,
|
|
message: Optional[str] = None
|
|
) -> "PaginatedResponse":
|
|
"""Create a paginated response with standard metadata."""
|
|
|
|
total_pages = (total + page_size - 1) // page_size
|
|
has_next = page < total_pages
|
|
has_prev = page > 1
|
|
|
|
pagination = {
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"total_pages": total_pages,
|
|
"has_next": has_next,
|
|
"has_prev": has_prev,
|
|
"next_page": page + 1 if has_next else None,
|
|
"prev_page": page - 1 if has_prev else None
|
|
}
|
|
|
|
return cls(
|
|
data=items,
|
|
pagination=pagination,
|
|
message=message
|
|
)
|
|
|
|
|
|
class BulkOperationResponse(SuccessResponse):
|
|
"""Response for bulk operations with detailed results."""
|
|
|
|
data: Dict[str, Any] = Field(..., description="Bulk operation results")
|
|
|
|
@classmethod
|
|
def create(
|
|
cls,
|
|
total_processed: int,
|
|
successful: int,
|
|
failed: int,
|
|
errors: Optional[List[ErrorDetail]] = None,
|
|
message: Optional[str] = None
|
|
) -> "BulkOperationResponse":
|
|
"""Create a bulk operation response with standard metadata."""
|
|
|
|
data = {
|
|
"total_processed": total_processed,
|
|
"successful": successful,
|
|
"failed": failed,
|
|
"success_rate": (successful / total_processed * 100) if total_processed > 0 else 0
|
|
}
|
|
|
|
if errors:
|
|
data["errors"] = [error.dict() for error in errors]
|
|
|
|
return cls(
|
|
data=data,
|
|
message=message or f"Processed {total_processed} items: {successful} successful, {failed} failed"
|
|
)
|
|
|
|
|
|
def create_error_response(
|
|
code: ErrorCode,
|
|
message: str,
|
|
field: Optional[str] = None,
|
|
context: Optional[Dict[str, Any]] = None,
|
|
request_id: Optional[str] = None
|
|
) -> ErrorResponse:
|
|
"""Helper function to create standardized error responses."""
|
|
|
|
error_detail = ErrorDetail(
|
|
code=code,
|
|
message=message,
|
|
field=field,
|
|
context=context
|
|
)
|
|
|
|
return ErrorResponse(
|
|
error=error_detail,
|
|
request_id=request_id
|
|
)
|
|
|
|
|
|
def create_validation_error_response(
|
|
field: str,
|
|
message: str,
|
|
input_value: Optional[Any] = None,
|
|
constraint: Optional[str] = None,
|
|
additional_errors: Optional[List[ValidationErrorDetail]] = None,
|
|
request_id: Optional[str] = None
|
|
) -> ValidationErrorResponse:
|
|
"""Helper function to create validation error responses."""
|
|
|
|
primary_error = ValidationErrorDetail(
|
|
code=ErrorCode.VALIDATION_ERROR,
|
|
message=message,
|
|
field=field,
|
|
input_value=input_value,
|
|
constraint=constraint
|
|
)
|
|
|
|
return ValidationErrorResponse(
|
|
error=primary_error,
|
|
errors=additional_errors,
|
|
request_id=request_id
|
|
)
|
|
|
|
|
|
def create_success_response(
|
|
data: Optional[Any] = None,
|
|
message: Optional[str] = None
|
|
) -> SuccessResponse:
|
|
"""Helper function to create standardized success responses."""
|
|
|
|
return SuccessResponse(
|
|
data=data,
|
|
message=message
|
|
)
|
|
|
|
|
|
def create_not_found_response(
|
|
resource: str,
|
|
identifier: Optional[str] = None,
|
|
request_id: Optional[str] = None
|
|
) -> ErrorResponse:
|
|
"""Helper function to create not found error responses."""
|
|
|
|
message = f"{resource} not found"
|
|
if identifier:
|
|
message += f" with identifier: {identifier}"
|
|
|
|
return create_error_response(
|
|
code=ErrorCode.NOT_FOUND,
|
|
message=message,
|
|
context={"resource": resource, "identifier": identifier},
|
|
request_id=request_id
|
|
)
|
|
|
|
|
|
def create_conflict_response(
|
|
resource: str,
|
|
reason: str,
|
|
request_id: Optional[str] = None
|
|
) -> ErrorResponse:
|
|
"""Helper function to create conflict error responses."""
|
|
|
|
return create_error_response(
|
|
code=ErrorCode.CONFLICT,
|
|
message=f"Conflict with {resource}: {reason}",
|
|
context={"resource": resource, "reason": reason},
|
|
request_id=request_id
|
|
)
|
|
|
|
|
|
def create_unauthorized_response(
|
|
reason: Optional[str] = None,
|
|
request_id: Optional[str] = None
|
|
) -> ErrorResponse:
|
|
"""Helper function to create unauthorized error responses."""
|
|
|
|
message = "Authentication required"
|
|
if reason:
|
|
message += f": {reason}"
|
|
|
|
return create_error_response(
|
|
code=ErrorCode.UNAUTHORIZED,
|
|
message=message,
|
|
context={"reason": reason},
|
|
request_id=request_id
|
|
)
|
|
|
|
|
|
def create_forbidden_response(
|
|
action: Optional[str] = None,
|
|
resource: Optional[str] = None,
|
|
request_id: Optional[str] = None
|
|
) -> ErrorResponse:
|
|
"""Helper function to create forbidden error responses."""
|
|
|
|
message = "Access denied"
|
|
context = {}
|
|
|
|
if action and resource:
|
|
message += f": insufficient permissions to {action} {resource}"
|
|
context = {"action": action, "resource": resource}
|
|
elif action:
|
|
message += f": insufficient permissions to {action}"
|
|
context = {"action": action}
|
|
elif resource:
|
|
message += f": insufficient permissions for {resource}"
|
|
context = {"resource": resource}
|
|
|
|
return create_error_response(
|
|
code=ErrorCode.FORBIDDEN,
|
|
message=message,
|
|
context=context,
|
|
request_id=request_id
|
|
)
|
|
|
|
|
|
# HTTP status code mapping for error codes
|
|
ERROR_CODE_STATUS_MAP = {
|
|
ErrorCode.VALIDATION_ERROR: 422,
|
|
ErrorCode.FIELD_REQUIRED: 422,
|
|
ErrorCode.FIELD_INVALID: 422,
|
|
ErrorCode.UNAUTHORIZED: 401,
|
|
ErrorCode.FORBIDDEN: 403,
|
|
ErrorCode.TOKEN_EXPIRED: 401,
|
|
ErrorCode.TOKEN_INVALID: 401,
|
|
ErrorCode.NOT_FOUND: 404,
|
|
ErrorCode.ALREADY_EXISTS: 409,
|
|
ErrorCode.CONFLICT: 409,
|
|
ErrorCode.DATABASE_ERROR: 500,
|
|
ErrorCode.INTEGRITY_ERROR: 400,
|
|
ErrorCode.TRANSACTION_ERROR: 500,
|
|
ErrorCode.BUSINESS_RULE_VIOLATION: 400,
|
|
ErrorCode.INVALID_OPERATION: 400,
|
|
ErrorCode.INTERNAL_ERROR: 500,
|
|
ErrorCode.SERVICE_UNAVAILABLE: 503,
|
|
ErrorCode.TIMEOUT: 504,
|
|
ErrorCode.FILE_TOO_LARGE: 413,
|
|
ErrorCode.INVALID_FILE_FORMAT: 400,
|
|
ErrorCode.IMPORT_ERROR: 400,
|
|
ErrorCode.SECURITY_VIOLATION: 403,
|
|
ErrorCode.RATE_LIMITED: 429,
|
|
}
|
|
|
|
|
|
def get_status_code_for_error(error_code: ErrorCode) -> int:
|
|
"""Get appropriate HTTP status code for an error code."""
|
|
return ERROR_CODE_STATUS_MAP.get(error_code, 500) |