Coverage for dataexcept/pipeline_exceptions.py: 100%

1"""Additional exception classes for data pipeline workflows.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any, Optional 

6 

7from .base import DataExceptError 

8from .redaction import redact_if_url, redact_url 

9 

10 

11class PipelineError(DataExceptError): 

12 """Base exception for pipeline errors.""" 

13 

14 pass 

15 

16 

17class PreprocessingError(PipelineError): 

18 """Raised when a preprocessing step fails.""" 

19 

20 def __init__(self, step_name: str, details: Optional[str] = None) -> None: 

21 default = f"Preprocessing failed at step: '{step_name}'." 

22 message = f"{default} Details: {details}" if details else default 

23 self.step_name = step_name 

24 self.details = details 

25 super().__init__(message) 

26 

27 

28class FeaturePreprocessingError(PreprocessingError): 

29 """Raised when feature engineering fails.""" 

30 

31 def __init__(self, feature: str, reason: Optional[str] = None) -> None: 

32 # Assigned before super(): DataExceptError.__init__ sweeps the stored 

33 # strings for URLs, and anything set afterwards escapes that. 

34 self.feature = feature 

35 self.reason = reason 

36 super().__init__(step_name=f"feature_{feature}", details=reason) 

37 

38 

39class StorageError(PipelineError): 

40 """Raised when reading from or writing to storage fails.""" 

41 

42 def __init__( 

43 self, 

44 location: str, 

45 operation: str, 

46 message: Optional[str] = None, 

47 ) -> None: 

48 default = f"Storage {operation} failed at location: '{location}'." 

49 self.location = redact_if_url(location) 

50 self.operation = operation 

51 super().__init__(message or default) 

52 

53 

54class PipelineNotificationError(PipelineError): 

55 """Raised when sending a notification fails.""" 

56 

57 def __init__( 

58 self, 

59 channel: str, 

60 payload: Any, 

61 message: Optional[str] = None, 

62 ) -> None: 

63 default = f"Notification via '{channel}' failed." 

64 self.channel = channel 

65 self.payload = payload 

66 super().__init__(message or default) 

67 

68 

69class RetryLimitExceededError(PipelineError): 

70 """Raised when an operation is retried too many times.""" 

71 

72 def __init__( 

73 self, 

74 operation: str, 

75 retries: int, 

76 message: Optional[str] = None, 

77 ) -> None: 

78 default = ( 

79 "Retry limit exceeded for operation " 

80 f"'{operation}' after {retries} attempts." 

81 ) 

82 self.operation = operation 

83 self.retries = retries 

84 super().__init__(message or default) 

85 

86 

87class ExternalServiceError(PipelineError): 

88 """General failure when calling an external service.""" 

89 

90 def __init__( 

91 self, 

92 service_name: str, 

93 status_code: Optional[int] = None, 

94 response: Optional[Any] = None, 

95 message: Optional[str] = None, 

96 ) -> None: 

97 default = f"Call to external service '{service_name}' failed." 

98 self.service_name = service_name 

99 self.status_code = status_code 

100 self.response = response 

101 super().__init__(message or default) 

102 

103 

104class ServiceAuthenticationError(ExternalServiceError): 

105 """Authentication to an external service failed.""" 

106 

107 def __init__( 

108 self, 

109 service_name: str, 

110 message: Optional[str] = None, 

111 ) -> None: 

112 default = f"Authentication failed for service '{service_name}'." 

113 super().__init__(service_name=service_name, message=message or default) 

114 

115 

116class ServiceAuthorizationError(ExternalServiceError): 

117 """Authorization was denied by an external service.""" 

118 

119 def __init__( 

120 self, 

121 service_name: str, 

122 message: Optional[str] = None, 

123 ) -> None: 

124 default = f"Authorization denied for service '{service_name}'." 

125 super().__init__(service_name=service_name, message=message or default) 

126 

127 

128class ServiceTimeoutError(ExternalServiceError): 

129 """A call to an external service exceeded the allotted time.""" 

130 

131 def __init__( 

132 self, 

133 service_name: str, 

134 timeout_seconds: Optional[float] = None, 

135 ) -> None: 

136 default = ( 

137 "Operation timed out after " 

138 f"{timeout_seconds}s on service '{service_name}'." 

139 ) 

140 self.timeout_seconds = timeout_seconds 

141 super().__init__(service_name=service_name, message=default) 

142 

143 

144class ApiError(PipelineError): 

145 """Failure calling a REST API endpoint.""" 

146 

147 def __init__( 

148 self, 

149 endpoint: str, 

150 status_code: Optional[int] = None, 

151 message: Optional[str] = None, 

152 ) -> None: 

153 # An endpoint URL may authenticate through a query parameter. 

154 self.endpoint = redact_url(endpoint) 

155 default = f"API call failed: {self.endpoint}" 

156 if status_code is not None: 

157 default += f" (status {status_code})" 

158 self.status_code = status_code 

159 super().__init__(message or default) 

160 

161 

162class TimeDeltaTooLargeError(PipelineError): 

163 """The time span between records exceeded a threshold.""" 

164 

165 def __init__( 

166 self, 

167 user: str, 

168 delta_minutes: float, 

169 message: Optional[str] = None, 

170 ) -> None: 

171 default = f"Time delta {delta_minutes}m too large for user {user}" 

172 self.user = user 

173 self.delta_minutes = delta_minutes 

174 super().__init__(message or default) 

175 

176 

177class TypeCheckError(PipelineError): 

178 """Invalid type detected during recursive type inspection.""" 

179 

180 

181class DataFetchError(PipelineError): 

182 """Failed to fetch data from a storage backend.""" 

183 

184 def __init__( 

185 self, 

186 source: str, 

187 cid: str, 

188 message: Optional[str] = None, 

189 ) -> None: 

190 default = f"Failed to fetch '{source}' data for cid={cid}" 

191 self.source = redact_if_url(source) 

192 self.cid = cid 

193 super().__init__(message or default) 

194 

195 

196__all__ = [ 

197 "PipelineError", 

198 "PreprocessingError", 

199 "FeaturePreprocessingError", 

200 "StorageError", 

201 "PipelineNotificationError", 

202 "RetryLimitExceededError", 

203 "ExternalServiceError", 

204 "ServiceAuthenticationError", 

205 "ServiceAuthorizationError", 

206 "ServiceTimeoutError", 

207 "ApiError", 

208 "TimeDeltaTooLargeError", 

209 "TypeCheckError", 

210 "DataFetchError", 

211]