- Add documentation warnings for in-memory rate limiting and failed login attempts - Consolidate duplicate health endpoints into api/health.py - Fix CLI to use correct async rescan method names - Update download.py and anime.py to use custom exception classes - Add WebSocket room validation and rate limiting
273 lines
7.1 KiB
Python
273 lines
7.1 KiB
Python
"""
|
|
Custom exception classes for Aniworld API layer.
|
|
|
|
This module defines exception hierarchy for the web API with proper
|
|
HTTP status code mappings and error handling.
|
|
"""
|
|
from typing import Any, Dict, Optional
|
|
|
|
|
|
class AniWorldAPIException(Exception):
|
|
"""
|
|
Base exception for Aniworld API.
|
|
|
|
All API-specific exceptions inherit from this class.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
status_code: int = 500,
|
|
error_code: Optional[str] = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""
|
|
Initialize API exception.
|
|
|
|
Args:
|
|
message: Human-readable error message
|
|
status_code: HTTP status code for response
|
|
error_code: Machine-readable error identifier
|
|
details: Additional error details and context
|
|
"""
|
|
self.message = message
|
|
self.status_code = status_code
|
|
self.error_code = error_code or self.__class__.__name__
|
|
self.details = details or {}
|
|
super().__init__(self.message)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""
|
|
Convert exception to dictionary for JSON response.
|
|
|
|
Returns:
|
|
Dictionary containing error information
|
|
"""
|
|
return {
|
|
"error": self.error_code,
|
|
"message": self.message,
|
|
"details": self.details,
|
|
}
|
|
|
|
|
|
class AuthenticationError(AniWorldAPIException):
|
|
"""Exception raised when authentication fails."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Authentication failed",
|
|
details: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""Initialize authentication error."""
|
|
super().__init__(
|
|
message=message,
|
|
status_code=401,
|
|
error_code="AUTHENTICATION_ERROR",
|
|
details=details,
|
|
)
|
|
|
|
|
|
class AuthorizationError(AniWorldAPIException):
|
|
"""Exception raised when user lacks required permissions."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Insufficient permissions",
|
|
details: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""Initialize authorization error."""
|
|
super().__init__(
|
|
message=message,
|
|
status_code=403,
|
|
error_code="AUTHORIZATION_ERROR",
|
|
details=details,
|
|
)
|
|
|
|
|
|
class ValidationError(AniWorldAPIException):
|
|
"""Exception raised when request validation fails."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Request validation failed",
|
|
details: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""Initialize validation error."""
|
|
super().__init__(
|
|
message=message,
|
|
status_code=422,
|
|
error_code="VALIDATION_ERROR",
|
|
details=details,
|
|
)
|
|
|
|
|
|
class NotFoundError(AniWorldAPIException):
|
|
"""Exception raised when resource is not found."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Resource not found",
|
|
resource_type: Optional[str] = None,
|
|
resource_id: Optional[Any] = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""Initialize not found error."""
|
|
if details is None:
|
|
details = {}
|
|
if resource_type:
|
|
details["resource_type"] = resource_type
|
|
if resource_id:
|
|
details["resource_id"] = resource_id
|
|
|
|
super().__init__(
|
|
message=message,
|
|
status_code=404,
|
|
error_code="NOT_FOUND",
|
|
details=details,
|
|
)
|
|
|
|
|
|
class ConflictError(AniWorldAPIException):
|
|
"""Exception raised when resource conflict occurs."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Resource conflict",
|
|
details: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""Initialize conflict error."""
|
|
super().__init__(
|
|
message=message,
|
|
status_code=409,
|
|
error_code="CONFLICT",
|
|
details=details,
|
|
)
|
|
|
|
|
|
class BadRequestError(AniWorldAPIException):
|
|
"""Exception raised for bad request (400) errors."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Bad request",
|
|
details: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""Initialize bad request error."""
|
|
super().__init__(
|
|
message=message,
|
|
status_code=400,
|
|
error_code="BAD_REQUEST",
|
|
details=details,
|
|
)
|
|
|
|
|
|
class RateLimitError(AniWorldAPIException):
|
|
"""Exception raised when rate limit is exceeded."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Rate limit exceeded",
|
|
retry_after: Optional[int] = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""Initialize rate limit error."""
|
|
if details is None:
|
|
details = {}
|
|
if retry_after:
|
|
details["retry_after"] = retry_after
|
|
|
|
super().__init__(
|
|
message=message,
|
|
status_code=429,
|
|
error_code="RATE_LIMIT_EXCEEDED",
|
|
details=details,
|
|
)
|
|
|
|
|
|
class ServerError(AniWorldAPIException):
|
|
"""Exception raised for internal server errors."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Internal server error",
|
|
error_code: str = "INTERNAL_SERVER_ERROR",
|
|
details: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""Initialize server error."""
|
|
super().__init__(
|
|
message=message,
|
|
status_code=500,
|
|
error_code=error_code,
|
|
details=details,
|
|
)
|
|
|
|
|
|
class DownloadError(ServerError):
|
|
"""Exception raised when download operation fails."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Download failed",
|
|
details: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""Initialize download error."""
|
|
super().__init__(
|
|
message=message,
|
|
error_code="DOWNLOAD_ERROR",
|
|
details=details,
|
|
)
|
|
|
|
|
|
class ConfigurationError(ServerError):
|
|
"""Exception raised when configuration is invalid."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Configuration error",
|
|
details: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""Initialize configuration error."""
|
|
super().__init__(
|
|
message=message,
|
|
error_code="CONFIGURATION_ERROR",
|
|
details=details,
|
|
)
|
|
|
|
|
|
class ProviderError(ServerError):
|
|
"""Exception raised when provider operation fails."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Provider error",
|
|
provider_name: Optional[str] = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""Initialize provider error."""
|
|
if details is None:
|
|
details = {}
|
|
if provider_name:
|
|
details["provider"] = provider_name
|
|
|
|
super().__init__(
|
|
message=message,
|
|
error_code="PROVIDER_ERROR",
|
|
details=details,
|
|
)
|
|
|
|
|
|
class DatabaseError(ServerError):
|
|
"""Exception raised when database operation fails."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Database error",
|
|
details: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""Initialize database error."""
|
|
super().__init__(
|
|
message=message,
|
|
error_code="DATABASE_ERROR",
|
|
details=details,
|
|
)
|