Medium-MCP / src /exceptions.py
Nikhil Pravin Pise
feat: implement comprehensive improvement plan (Phases 1-5)
e98cc10
"""
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)