""" Custom Exceptions Module Defines structured exceptions for better error handling and API responses. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Optional from enum import Enum # ============================================================================= # ERROR CODES # ============================================================================= class ErrorCode(str, Enum): """Standard error codes for API responses.""" # Validation errors (400) INVALID_URL = "INVALID_URL" INVALID_QUERY = "INVALID_QUERY" INVALID_TAG = "INVALID_TAG" INVALID_FORMAT = "INVALID_FORMAT" VALIDATION_ERROR = "VALIDATION_ERROR" # Authentication/Authorization (401/403) UNAUTHORIZED = "UNAUTHORIZED" FORBIDDEN = "FORBIDDEN" API_KEY_MISSING = "API_KEY_MISSING" API_KEY_INVALID = "API_KEY_INVALID" # Not found (404) ARTICLE_NOT_FOUND = "ARTICLE_NOT_FOUND" RESOURCE_NOT_FOUND = "RESOURCE_NOT_FOUND" # Rate limiting (429) RATE_LIMITED = "RATE_LIMITED" QUOTA_EXCEEDED = "QUOTA_EXCEEDED" # Server errors (500) INTERNAL_ERROR = "INTERNAL_ERROR" SCRAPE_FAILED = "SCRAPE_FAILED" EXTRACTION_FAILED = "EXTRACTION_FAILED" RENDER_FAILED = "RENDER_FAILED" # External service errors (502/503) MEDIUM_UNAVAILABLE = "MEDIUM_UNAVAILABLE" SERVICE_UNAVAILABLE = "SERVICE_UNAVAILABLE" EXTERNAL_API_ERROR = "EXTERNAL_API_ERROR" # Timeout (504) TIMEOUT = "TIMEOUT" # ============================================================================= # ERROR RESPONSE # ============================================================================= @dataclass class ErrorResponse: """Structured error response for API/UI.""" code: ErrorCode message: str details: Optional[dict[str, Any]] = None http_status: int = 400 recoverable: bool = True def to_dict(self) -> dict[str, Any]: """Convert to dictionary for JSON serialization.""" result = { "error": { "code": self.code.value, "message": self.message, } } if self.details: result["error"]["details"] = self.details return result def to_user_message(self) -> str: """Get user-friendly error message.""" return self.message # ============================================================================= # BASE EXCEPTION # ============================================================================= class MediumMCPError(Exception): """ Base exception for all Medium-MCP errors. All custom exceptions should inherit from this class. """ code: ErrorCode = ErrorCode.INTERNAL_ERROR http_status: int = 500 recoverable: bool = True def __init__( self, message: str, details: Optional[dict[str, Any]] = None, cause: Optional[Exception] = None, ) -> None: super().__init__(message) self.message = message self.details = details or {} self.cause = cause def to_response(self) -> ErrorResponse: """Convert exception to ErrorResponse.""" return ErrorResponse( code=self.code, message=self.message, details=self.details, http_status=self.http_status, recoverable=self.recoverable, ) # ============================================================================= # VALIDATION EXCEPTIONS # ============================================================================= class ValidationError(MediumMCPError): """Raised when input validation fails.""" code = ErrorCode.VALIDATION_ERROR http_status = 400 class InvalidURLError(ValidationError): """Raised when URL is invalid.""" code = ErrorCode.INVALID_URL def __init__(self, url: str, reason: str = "Invalid URL format") -> None: super().__init__( message=f"Invalid URL: {reason}", details={"url": url[:100], "reason": reason}, ) class InvalidQueryError(ValidationError): """Raised when search query is invalid.""" code = ErrorCode.INVALID_QUERY def __init__(self, query: str, reason: str = "Invalid query") -> None: super().__init__( message=f"Invalid search query: {reason}", details={"query": query[:100], "reason": reason}, ) class InvalidTagError(ValidationError): """Raised when tag is invalid.""" code = ErrorCode.INVALID_TAG def __init__(self, tag: str, reason: str = "Invalid tag format") -> None: super().__init__( message=f"Invalid tag: {reason}", details={"tag": tag[:50], "reason": reason}, ) # ============================================================================= # SCRAPING EXCEPTIONS # ============================================================================= class ScrapeError(MediumMCPError): """Base exception for scraping failures.""" code = ErrorCode.SCRAPE_FAILED http_status = 502 class ArticleNotFoundError(ScrapeError): """Raised when article cannot be found.""" code = ErrorCode.ARTICLE_NOT_FOUND http_status = 404 def __init__(self, url: str) -> None: super().__init__( message=f"Article not found: {url}", details={"url": url}, ) class ExtractionError(ScrapeError): """Raised when content extraction fails.""" code = ErrorCode.EXTRACTION_FAILED def __init__(self, url: str, tier: str = "unknown") -> None: super().__init__( message=f"Failed to extract content from {url}", details={"url": url, "tier": tier}, ) class PaywallError(ScrapeError): """Raised when article is paywalled.""" code = ErrorCode.SCRAPE_FAILED recoverable = False def __init__(self, url: str) -> None: super().__init__( message="Article is behind a paywall", details={"url": url, "paywalled": True}, ) # ============================================================================= # RATE LIMITING EXCEPTIONS # ============================================================================= class RateLimitError(MediumMCPError): """Raised when rate limit is exceeded.""" code = ErrorCode.RATE_LIMITED http_status = 429 def __init__( self, message: str = "Rate limit exceeded", retry_after: Optional[int] = None, ) -> None: details = {} if retry_after: details["retry_after_seconds"] = retry_after super().__init__(message=message, details=details) self.retry_after = retry_after class QuotaExceededError(RateLimitError): """Raised when API quota is exceeded.""" code = ErrorCode.QUOTA_EXCEEDED recoverable = False def __init__(self, service: str = "Unknown") -> None: super().__init__( message=f"API quota exceeded for {service}", ) # ============================================================================= # EXTERNAL SERVICE EXCEPTIONS # ============================================================================= class ExternalServiceError(MediumMCPError): """Raised when an external service fails.""" code = ErrorCode.EXTERNAL_API_ERROR http_status = 502 def __init__( self, service: str, message: str = "External service error", status_code: Optional[int] = None, ) -> None: details = {"service": service} if status_code: details["status_code"] = status_code super().__init__(message=f"{service}: {message}", details=details) class MediumUnavailableError(ExternalServiceError): """Raised when Medium is unavailable.""" code = ErrorCode.MEDIUM_UNAVAILABLE http_status = 503 def __init__(self, message: str = "Medium is currently unavailable") -> None: super().__init__(service="Medium", message=message) class TimeoutError(MediumMCPError): """Raised when an operation times out.""" code = ErrorCode.TIMEOUT http_status = 504 def __init__( self, operation: str = "request", timeout_seconds: Optional[int] = None, ) -> None: details = {"operation": operation} if timeout_seconds: details["timeout_seconds"] = timeout_seconds super().__init__( message=f"Operation timed out: {operation}", details=details, ) # ============================================================================= # RENDER EXCEPTIONS # ============================================================================= class RenderError(MediumMCPError): """Raised when rendering fails.""" code = ErrorCode.RENDER_FAILED http_status = 500 def __init__( self, format: str = "unknown", message: str = "Render failed", ) -> None: super().__init__( message=f"Failed to render {format}: {message}", details={"format": format}, ) class PDFRenderError(RenderError): """Raised when PDF rendering fails.""" def __init__(self, message: str = "PDF generation failed") -> None: super().__init__(format="PDF", message=message)