kami_spider_monorepo/core/exceptions.py

"""
Custom exception hierarchy for application errors.
Maps exceptions to HTTP status codes and business codes.
"""

from typing import Optional, Any
from core.responses import BusinessCode


class BaseAppException(Exception):
    """
    Base exception for all application exceptions.

    Attributes:
        message: Error message
        code: Business error code
        status_code: HTTP status code
        details: Additional error details
    """

    def __init__(
        self,
        message: str,
        code: BusinessCode = BusinessCode.UNKNOWN_ERROR,
        status_code: int = 500,
        details: Optional[dict[str, Any]] = None,
    ):
        self.message = message
        self.code = code
        self.status_code = status_code
        self.details = details or {}
        super().__init__(self.message)


class ValidationException(BaseAppException):
    """
    Exception for validation errors.

    HTTP Status: 400 Bad Request
    Business Code Range: 3000-3999
    """

    def __init__(
        self,
        message: str,
        code: BusinessCode = BusinessCode.INVALID_INPUT,
        details: Optional[dict[str, Any]] = None,
    ):
        super().__init__(message=message, code=code, status_code=400, details=details)


class NotFoundException(BaseAppException):
    """
    Exception for resource not found errors.

    HTTP Status: 404 Not Found
    Business Code: 4001
    """

    def __init__(
        self,
        message: str,
        resource: str = "Resource",
        details: Optional[dict[str, Any]] = None,
    ):
        super().__init__(
            message=message or f"{resource} not found",
            code=BusinessCode.RESOURCE_NOT_FOUND,
            status_code=404,
            details=details,
        )


class ConflictException(BaseAppException):
    """
    Exception for resource conflict errors.

    HTTP Status: 409 Conflict
    Business Code: 4003
    """

    def __init__(
        self,
        message: str,
        code: BusinessCode = BusinessCode.RESOURCE_CONFLICT,
        details: Optional[dict[str, Any]] = None,
    ):
        super().__init__(message=message, code=code, status_code=409, details=details)


class AlreadyExistsException(BaseAppException):
    """
    Exception for resource already exists errors.

    HTTP Status: 409 Conflict
    Business Code: 4002
    """

    def __init__(
        self,
        message: str,
        resource: str = "Resource",
        details: Optional[dict[str, Any]] = None,
    ):
        super().__init__(
            message=message or f"{resource} already exists",
            code=BusinessCode.RESOURCE_ALREADY_EXISTS,
            status_code=409,
            details=details,
        )


class AuthenticationException(BaseAppException):
    """
    Exception for authentication errors.

    HTTP Status: 401 Unauthorized
    Business Code Range: 1001-1099
    """

    def __init__(
        self,
        message: str,
        code: BusinessCode = BusinessCode.LOGIN_FAILED,
        details: Optional[dict[str, Any]] = None,
    ):
        super().__init__(message=message, code=code, status_code=401, details=details)


class PermissionException(BaseAppException):
    """
    Exception for permission/authorization errors.

    HTTP Status: 403 Forbidden
    Business Code: 1003
    """

    def __init__(
        self,
        message: str,
        code: BusinessCode = BusinessCode.INSUFFICIENT_PERMISSIONS,
        details: Optional[dict[str, Any]] = None,
    ):
        super().__init__(message=message, code=code, status_code=403, details=details)


class BusinessLogicException(BaseAppException):
    """
    Exception for business logic errors.

    HTTP Status: 400 Bad Request
    Business Code Range: 2000-2999
    """

    def __init__(
        self,
        message: str,
        code: BusinessCode = BusinessCode.OPERATION_NOT_ALLOWED,
        details: Optional[dict[str, Any]] = None,
    ):
        super().__init__(message=message, code=code, status_code=400, details=details)


class DatabaseException(BaseAppException):
    """
    Exception for database errors.

    HTTP Status: 503 Service Unavailable
    Business Code: 5001
    """

    def __init__(
        self,
        message: str = "Database error occurred",
        details: Optional[dict[str, Any]] = None,
    ):
        super().__init__(
            message=message,
            code=BusinessCode.DATABASE_ERROR,
            status_code=503,
            details=details,
        )


class CacheException(BaseAppException):
    """
    Exception for cache/Redis errors.

    HTTP Status: 503 Service Unavailable
    Business Code: 5003
    """

    def __init__(
        self,
        message: str = "Cache service error",
        details: Optional[dict[str, Any]] = None,
    ):
        super().__init__(
            message=message,
            code=BusinessCode.CACHE_ERROR,
            status_code=503,
            details=details,
        )


class ExternalServiceException(BaseAppException):
    """
    Exception for external service errors.

    HTTP Status: 502 Bad Gateway
    Business Code: 5002
    """

    def __init__(
        self,
        message: str = "External service unavailable",
        details: Optional[dict[str, Any]] = None,
    ):
        super().__init__(
            message=message,
            code=BusinessCode.EXTERNAL_SERVICE_ERROR,
            status_code=502,
            details=details,
        )


class JDServiceException(BaseAppException):
    """
    Exception for external service errors.

    HTTP Status: 502 Bad Gateway
    Business Code: 5002
    """

    def __init__(
        self,
        code: BusinessCode = BusinessCode.NOT_IMPLEMENTED,
        message: str = "External service unavailable",
        details: Optional[dict[str, Any]] = None,
    ):
        super().__init__(
            message=message,
            code=code,
            status_code=200,
            details=details,
        )