Coverage for src / local_deep_research / news / exceptions.py: 97%
55 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-01-11 00:51 +0000
« prev ^ index » next coverage.py v7.12.0, created at 2026-01-11 00:51 +0000
1"""
2Custom exceptions for the news API module.
4These exceptions are used to provide structured error handling
5that can be caught by Flask error handlers and converted to
6appropriate JSON responses.
7"""
9from typing import Optional, Dict, Any
12class NewsAPIException(Exception):
13 """Base exception for all news API related errors."""
15 def __init__(
16 self,
17 message: str,
18 status_code: int = 500,
19 error_code: Optional[str] = None,
20 details: Optional[Dict[str, Any]] = None,
21 ):
22 """
23 Initialize the news API exception.
25 Args:
26 message: Human-readable error message
27 status_code: HTTP status code for the error
28 error_code: Machine-readable error code for API consumers
29 details: Additional error details/context
30 """
31 super().__init__(message)
32 self.message = message
33 self.status_code = status_code
34 self.error_code = error_code or self.__class__.__name__
35 self.details = details or {}
37 def to_dict(self) -> Dict[str, Any]:
38 """Convert exception to dictionary for JSON response."""
39 result = {
40 "error": self.message,
41 "error_code": self.error_code,
42 "status_code": self.status_code,
43 }
44 if self.details:
45 result["details"] = self.details
46 return result
49class NewsFeatureDisabledException(NewsAPIException):
50 """Raised when the news feature is disabled in settings."""
52 def __init__(self, message: str = "News system is disabled"):
53 super().__init__(message, status_code=503, error_code="NEWS_DISABLED")
56class InvalidLimitException(NewsAPIException):
57 """Raised when an invalid limit parameter is provided."""
59 def __init__(self, limit: int):
60 super().__init__(
61 f"Invalid limit: {limit}. Limit must be at least 1",
62 status_code=400,
63 error_code="INVALID_LIMIT",
64 details={"provided_limit": limit, "min_limit": 1},
65 )
68class SubscriptionNotFoundException(NewsAPIException):
69 """Raised when a requested subscription is not found."""
71 def __init__(self, subscription_id: str):
72 super().__init__(
73 f"Subscription not found: {subscription_id}",
74 status_code=404,
75 error_code="SUBSCRIPTION_NOT_FOUND",
76 details={"subscription_id": subscription_id},
77 )
80class SubscriptionCreationException(NewsAPIException):
81 """Raised when subscription creation fails."""
83 def __init__(self, message: str, details: Optional[Dict[str, Any]] = None):
84 super().__init__(
85 f"Failed to create subscription: {message}",
86 status_code=500,
87 error_code="SUBSCRIPTION_CREATE_FAILED",
88 details=details,
89 )
92class SubscriptionUpdateException(NewsAPIException):
93 """Raised when subscription update fails."""
95 def __init__(self, subscription_id: str, message: str):
96 super().__init__(
97 f"Failed to update subscription {subscription_id}: {message}",
98 status_code=500,
99 error_code="SUBSCRIPTION_UPDATE_FAILED",
100 details={"subscription_id": subscription_id},
101 )
104class SubscriptionDeletionException(NewsAPIException):
105 """Raised when subscription deletion fails."""
107 def __init__(self, subscription_id: str, message: str):
108 super().__init__(
109 f"Failed to delete subscription {subscription_id}: {message}",
110 status_code=500,
111 error_code="SUBSCRIPTION_DELETE_FAILED",
112 details={"subscription_id": subscription_id},
113 )
116class DatabaseAccessException(NewsAPIException):
117 """Raised when database access fails."""
119 def __init__(self, operation: str, message: str):
120 super().__init__(
121 f"Database error during {operation}: {message}",
122 status_code=500,
123 error_code="DATABASE_ERROR",
124 details={"operation": operation},
125 )
128class NewsFeedGenerationException(NewsAPIException):
129 """Raised when news feed generation fails."""
131 def __init__(self, message: str, user_id: Optional[str] = None):
132 details = {}
133 if user_id:
134 details["user_id"] = user_id
135 super().__init__(
136 f"Failed to generate news feed: {message}",
137 status_code=500,
138 error_code="FEED_GENERATION_FAILED",
139 details=details,
140 )
143class ResearchProcessingException(NewsAPIException):
144 """Raised when processing research items for news fails."""
146 def __init__(self, message: str, research_id: Optional[str] = None):
147 details = {}
148 if research_id: 148 ↛ 150line 148 didn't jump to line 150 because the condition on line 148 was always true
149 details["research_id"] = research_id
150 super().__init__(
151 f"Failed to process research item: {message}",
152 status_code=500,
153 error_code="RESEARCH_PROCESSING_FAILED",
154 details=details,
155 )
158class NotImplementedException(NewsAPIException):
159 """Raised when a feature is not yet implemented."""
161 def __init__(self, feature: str):
162 super().__init__(
163 f"Feature not yet implemented: {feature}",
164 status_code=501,
165 error_code="NOT_IMPLEMENTED",
166 details={"feature": feature},
167 )
170class InvalidParameterException(NewsAPIException):
171 """Raised when invalid parameters are provided to API functions."""
173 def __init__(self, parameter: str, value: Any, message: str):
174 super().__init__(
175 f"Invalid parameter '{parameter}': {message}",
176 status_code=400,
177 error_code="INVALID_PARAMETER",
178 details={"parameter": parameter, "value": value},
179 )
182class SchedulerNotificationException(NewsAPIException):
183 """Raised when scheduler notification fails (non-critical)."""
185 def __init__(self, action: str, message: str):
186 super().__init__(
187 f"Failed to notify scheduler about {action}: {message}",
188 status_code=500,
189 error_code="SCHEDULER_NOTIFICATION_FAILED",
190 details={"action": action},
191 )