Coverage for netrun_errors / service.py: 100%

1""" 

2Service-related exceptions for Netrun Systems. 

3 

4Exceptions for service availability, external dependencies, 

5and system-level errors. 

6""" 

7 

8from typing import Any, Dict, Optional 

9 

10from fastapi import status 

11 

12from .base import NetrunException 

13 

14 

15class ServiceUnavailableError(NetrunException): 

16 """ 

17 Raised when service or dependency is temporarily unavailable. 

18 

19 Common scenarios: 

20 - Database connection failures 

21 - External API timeouts 

22 - Maintenance mode 

23 - Circuit breaker open 

24 

25 Status Code: 503 Service Unavailable 

26 Error Code: SERVICE_UNAVAILABLE 

27 """ 

28 

29 def __init__( 

30 self, 

31 message: str = "Service is temporarily unavailable. Please try again later.", 

32 details: Optional[Dict[str, Any]] = None, 

33 correlation_id: Optional[str] = None, 

34 ): 

35 super().__init__( 

36 status_code=status.HTTP_503_SERVICE_UNAVAILABLE, 

37 error_code="SERVICE_UNAVAILABLE", 

38 message=message, 

39 details=details, 

40 correlation_id=correlation_id, 

41 ) 

42 

43 

44class TemporalUnavailableError(NetrunException): 

45 """ 

46 Raised when Temporal workflow engine is unavailable. 

47 

48 Specific to Netrun Systems' Temporal.io integration for 

49 long-running workflows and distributed transactions. 

50 

51 Status Code: 503 Service Unavailable 

52 Error Code: TEMPORAL_UNAVAILABLE 

53 """ 

54 

55 def __init__( 

56 self, 

57 message: str = "Workflow engine is temporarily unavailable. Please try again later.", 

58 details: Optional[Dict[str, Any]] = None, 

59 correlation_id: Optional[str] = None, 

60 ): 

61 super().__init__( 

62 status_code=status.HTTP_503_SERVICE_UNAVAILABLE, 

63 error_code="TEMPORAL_UNAVAILABLE", 

64 message=message, 

65 details=details, 

66 correlation_id=correlation_id, 

67 ) 

68 

69 

70class RateLimitExceededError(NetrunException): 

71 """ 

72 Raised when rate limit is exceeded. 

73 

74 Common scenarios: 

75 - API request quota exceeded 

76 - Too many login attempts 

77 - Bulk operation throttling 

78 

79 Status Code: 429 Too Many Requests 

80 Error Code: RATE_LIMIT_EXCEEDED 

81 """ 

82 

83 def __init__( 

84 self, 

85 message: str = "Rate limit exceeded. Please try again later.", 

86 retry_after: Optional[int] = None, 

87 details: Optional[Dict[str, Any]] = None, 

88 correlation_id: Optional[str] = None, 

89 ): 

90 details = details or {} 

91 if retry_after is not None: 

92 details["retry_after_seconds"] = retry_after 

93 

94 super().__init__( 

95 status_code=status.HTTP_429_TOO_MANY_REQUESTS, 

96 error_code="RATE_LIMIT_EXCEEDED", 

97 message=message, 

98 details=details, 

99 correlation_id=correlation_id, 

100 ) 

101 self.retry_after = retry_after 

102 

103 

104class BadGatewayError(NetrunException): 

105 """ 

106 Raised when upstream service returns invalid response. 

107 

108 Common scenarios: 

109 - Invalid response from upstream API 

110 - Proxy/load balancer errors 

111 - Malformed upstream responses 

112 

113 Status Code: 502 Bad Gateway 

114 Error Code: BAD_GATEWAY 

115 """ 

116 

117 def __init__( 

118 self, 

119 message: str = "Upstream service returned an invalid response.", 

120 upstream_service: Optional[str] = None, 

121 details: Optional[Dict[str, Any]] = None, 

122 correlation_id: Optional[str] = None, 

123 ): 

124 details = details or {} 

125 if upstream_service: 

126 details["upstream_service"] = upstream_service 

127 

128 super().__init__( 

129 status_code=status.HTTP_502_BAD_GATEWAY, 

130 error_code="BAD_GATEWAY", 

131 message=message, 

132 details=details, 

133 correlation_id=correlation_id, 

134 ) 

135 

136 

137class GatewayTimeoutError(NetrunException): 

138 """ 

139 Raised when upstream service times out. 

140 

141 Common scenarios: 

142 - Upstream API request timeout 

143 - Database query timeout 

144 - External service not responding 

145 

146 Status Code: 504 Gateway Timeout 

147 Error Code: GATEWAY_TIMEOUT 

148 """ 

149 

150 def __init__( 

151 self, 

152 message: str = "Upstream service timed out. Please try again.", 

153 upstream_service: Optional[str] = None, 

154 timeout_seconds: Optional[float] = None, 

155 details: Optional[Dict[str, Any]] = None, 

156 correlation_id: Optional[str] = None, 

157 ): 

158 details = details or {} 

159 if upstream_service: 

160 details["upstream_service"] = upstream_service 

161 if timeout_seconds is not None: 

162 details["timeout_seconds"] = timeout_seconds 

163 

164 super().__init__( 

165 status_code=status.HTTP_504_GATEWAY_TIMEOUT, 

166 error_code="GATEWAY_TIMEOUT", 

167 message=message, 

168 details=details, 

169 correlation_id=correlation_id, 

170 ) 

171 

172 

173class ExternalServiceError(NetrunException): 

174 """ 

175 Raised when external service integration fails. 

176 

177 Generic error for external API failures that don't fit 

178 specific categories (BadGateway, GatewayTimeout, etc.) 

179 

180 Status Code: 502 Bad Gateway 

181 Error Code: EXTERNAL_SERVICE_ERROR 

182 """ 

183 

184 def __init__( 

185 self, 

186 service_name: str, 

187 message: Optional[str] = None, 

188 original_error: Optional[str] = None, 

189 details: Optional[Dict[str, Any]] = None, 

190 correlation_id: Optional[str] = None, 

191 ): 

192 details = details or {} 

193 details["service_name"] = service_name 

194 if original_error: 

195 details["original_error"] = original_error 

196 

197 super().__init__( 

198 status_code=status.HTTP_502_BAD_GATEWAY, 

199 error_code="EXTERNAL_SERVICE_ERROR", 

200 message=message or f"External service '{service_name}' is unavailable.", 

201 details=details, 

202 correlation_id=correlation_id, 

203 )