Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |
| # ============================================================================= | |
| 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) | |