Coverage for dataexcept/datascience_exceptions/ingestion.py: 91%

1"""Data ingestion and validation related errors.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any, Optional, Sequence 

6 

7from .._validation import is_number 

8from ..redaction import redact_if_url 

9from .base import DataScienceError 

10 

11 

12class DataLoadingError(DataScienceError): 

13 """ 

14 Raised when loading data fails. 

15 

16 Attributes: 

17 source: data source description (file path, URL). 

18 original: underlying exception. 

19 """ 

20 

21 def __init__(self, source: str, original: Exception) -> None: 

22 if not isinstance(source, str): 

23 raise TypeError(f"source must be str, got {type(source).__name__}") 

24 if not isinstance(original, Exception): 

25 raise TypeError( 

26 f"original must be Exception, got {type(original).__name__}" 

27 ) 

28 

29 message = f"Failed to load data from {source!r}: {original}" 

30 self.source = redact_if_url(source) 

31 self.original = original 

32 super().__init__(message) 

33 

34 def __str__(self) -> str: 

35 return f"[DataLoadingError:{self.source}] {self.message}" 

36 

37 

38class DataFormatError(DataScienceError): 

39 """Raised when input data is not in the expected format.""" 

40 

41 def __init__(self, expected_formats: Sequence[str], found_format: str) -> None: 

42 if not isinstance(found_format, str): 

43 raise TypeError( 

44 f"found_format must be str, got {type(found_format).__name__}" 

45 ) 

46 if not isinstance(expected_formats, Sequence) or isinstance( 

47 expected_formats, str 

48 ): 

49 raise TypeError("expected_formats must be a sequence of strings") 

50 if not all(isinstance(fmt, str) for fmt in expected_formats): 

51 raise TypeError("expected_formats must contain strings") 

52 

53 self.expected_formats = list(expected_formats) 

54 self.found_format = found_format 

55 fmt_list = ", ".join(self.expected_formats) 

56 message = f"Expected data format {fmt_list}; got {found_format}" 

57 super().__init__(message) 

58 

59 def __str__(self) -> str: 

60 return f"[DataFormatError] {self.message}" 

61 

62 

63class DataValidationError(DataScienceError): 

64 """ 

65 Raised when data fails validation rules. 

66 

67 Attributes: 

68 field: name of invalid field. 

69 value: the invalid value. 

70 """ 

71 

72 def __init__(self, field: str, value: Any, message: Optional[str] = None) -> None: 

73 if not isinstance(field, str): 73 ↛ 74line 73 didn't jump to line 74 because the condition on line 73 was never true

74 raise TypeError(f"field must be str, got {type(field).__name__}") 

75 

76 if message is None: 

77 message = f"Invalid value for '{field}': {value!r}" 

78 elif not isinstance(message, str): 

79 raise TypeError(f"message must be str, got {type(message).__name__}") 

80 

81 self.field = field 

82 self.value = value 

83 super().__init__(message) 

84 

85 def __str__(self) -> str: 

86 return f"[DataValidationError:{self.field}] {self.message}" 

87 

88 

89class MissingDataError(DataScienceError): 

90 """ 

91 Raised when required data is missing. 

92 

93 Attributes: 

94 feature: name of missing feature. 

95 """ 

96 

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

98 if not isinstance(feature, str): 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true

99 raise TypeError(f"feature must be str, got {type(feature).__name__}") 

100 

101 if message is None: 

102 message = f"Missing required feature: {feature!r}" 

103 elif not isinstance(message, str): 103 ↛ 106line 103 didn't jump to line 106 because the condition on line 103 was always true

104 raise TypeError(f"message must be str, got {type(message).__name__}") 

105 

106 self.feature = feature 

107 super().__init__(message) 

108 

109 def __str__(self) -> str: 

110 return f"[MissingDataError:{self.feature}] {self.message}" 

111 

112 

113class OutlierDetectionError(DataScienceError): 

114 """ 

115 Raised when outlier detection fails. 

116 

117 Attributes: 

118 method: detection method name. 

119 details: optional extra info. 

120 """ 

121 

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

123 if not isinstance(method, str): 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true

124 raise TypeError(f"method must be str, got {type(method).__name__}") 

125 if details is not None and not isinstance(details, str): 

126 raise TypeError( 

127 f"details must be str or None, got {type(details).__name__}" 

128 ) 

129 

130 msg = f"Outlier detection failed using method '{method}'" 

131 if details: 

132 msg += f": {details}" 

133 

134 self.method = method 

135 self.details = details 

136 super().__init__(msg) 

137 

138 def __str__(self) -> str: 

139 return f"[OutlierDetectionError:{self.method}] {self.message}" 

140 

141 

142class SchemaMismatchError(DataScienceError): 

143 """ 

144 Raised when data schema does not match expected. 

145 

146 Attributes: 

147 expected: expected schema description. 

148 found: actual schema description. 

149 """ 

150 

151 def __init__(self, expected: str, found: str) -> None: 

152 if not isinstance(expected, str): 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true

153 raise TypeError(f"expected must be str, got {type(expected).__name__}") 

154 if not isinstance(found, str): 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true

155 raise TypeError(f"found must be str, got {type(found).__name__}") 

156 

157 message = f"Schema mismatch. Expected: {expected}, Found: {found}" 

158 self.expected = expected 

159 self.found = found 

160 super().__init__(message) 

161 

162 def __str__(self) -> str: 

163 return f"[SchemaMismatchError] {self.message}" 

164 

165 

166class FeatureEngineeringError(DataScienceError): 

167 """ 

168 Raised during feature engineering steps. 

169 

170 Attributes: 

171 step: description of the step that failed. 

172 cause: optional underlying reason. 

173 """ 

174 

175 def __init__(self, step: str, cause: Optional[str] = None) -> None: 

176 if not isinstance(step, str): 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true

177 raise TypeError(f"step must be str, got {type(step).__name__}") 

178 if cause is not None and not isinstance(cause, str): 178 ↛ 179line 178 didn't jump to line 179 because the condition on line 178 was never true

179 raise TypeError(f"cause must be str or None, got {type(cause).__name__}") 

180 

181 msg = f"Feature engineering failed at step '{step}'" 

182 if cause: 182 ↛ 183line 182 didn't jump to line 183 because the condition on line 182 was never true

183 msg += f": {cause}" 

184 

185 self.step = step 

186 self.cause = cause 

187 super().__init__(msg) 

188 

189 def __str__(self) -> str: 

190 return f"[FeatureEngineeringError] {self.message}" 

191 

192 

193class DataNormalizationError(DataScienceError): 

194 """Raised when data normalization fails. 

195 

196 Args: 

197 method: Normalization technique identifier. 

198 details: Optional explanation of the failure. 

199 """ 

200 

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

202 if not isinstance(method, str): 

203 raise TypeError(f"method must be str, got {type(method).__name__}") 

204 if details is not None and not isinstance(details, str): 

205 raise TypeError( 

206 f"details must be str or None, got {type(details).__name__}" 

207 ) 

208 # Build a helpful error message 

209 msg = f"Normalization using '{method}' failed" 

210 if details: 

211 msg += f": {details}" 

212 self.method = method 

213 self.details = details 

214 super().__init__(msg) 

215 

216 def __str__(self) -> str: 

217 return f"[DataNormalizationError:{self.method}] {self.message}" 

218 

219 

220class DataImbalanceError(DataScienceError): 

221 """Raised when class distribution is too imbalanced. 

222 

223 Args: 

224 ratio: Observed minority-to-majority ratio. 

225 threshold: Minimum acceptable ratio. 

226 message: Optional custom error message. 

227 """ 

228 

229 def __init__( 

230 self, ratio: float, threshold: float, message: Optional[str] = None 

231 ) -> None: 

232 if not is_number(ratio): 

233 raise TypeError(f"ratio must be numeric, got {type(ratio).__name__}") 

234 if not is_number(threshold): 

235 raise TypeError( 

236 f"threshold must be numeric, got {type(threshold).__name__}" 

237 ) 

238 if message is not None and not isinstance(message, str): 

239 raise TypeError( 

240 f"message must be str or None, got {type(message).__name__}" 

241 ) 

242 self.ratio = float(ratio) 

243 self.threshold = float(threshold) 

244 if message is None: 244 ↛ 250line 244 didn't jump to line 250 because the condition on line 244 was always true

245 msg = ( 

246 f"Data imbalance detected: ratio={self.ratio:.3f} < " 

247 f"threshold={self.threshold:.3f}" 

248 ) 

249 else: 

250 msg = message 

251 super().__init__(msg) 

252 

253 def __str__(self) -> str: 

254 return f"[DataImbalanceError] {self.message}" 

255 

256 

257class DataAugmentationError(DataScienceError): 

258 """Raised when a data augmentation technique fails. 

259 

260 Args: 

261 technique: Name of the augmentation technique. 

262 details: Optional explanation of the failure. 

263 """ 

264 

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

266 if not isinstance(technique, str): 

267 raise TypeError(f"technique must be str, got {type(technique).__name__}") 

268 if details is not None and not isinstance(details, str): 

269 raise TypeError( 

270 f"details must be str or None, got {type(details).__name__}" 

271 ) 

272 

273 msg = f"Data augmentation '{technique}' failed" 

274 if details: 

275 msg += f": {details}" 

276 

277 self.technique = technique 

278 self.details = details 

279 super().__init__(msg) 

280 

281 def __str__(self) -> str: 

282 return f"[DataAugmentationError:{self.technique}] {self.message}" 

283 

284 

285class DataLeakageError(DataScienceError): 

286 """Raised when data leakage is detected between train and test sets. 

287 

288 Args: 

289 feature: Name of the leaked feature. 

290 stage: Stage where the leakage occurred. 

291 message: Optional custom message. 

292 """ 

293 

294 def __init__(self, feature: str, stage: str, message: Optional[str] = None) -> None: 

295 if not isinstance(feature, str): 

296 raise TypeError(f"feature must be str, got {type(feature).__name__}") 

297 if not isinstance(stage, str): 

298 raise TypeError(f"stage must be str, got {type(stage).__name__}") 

299 if message is not None and not isinstance(message, str): 

300 raise TypeError( 

301 f"message must be str or None, got {type(message).__name__}" 

302 ) 

303 

304 if message is None: 304 ↛ 307line 304 didn't jump to line 307 because the condition on line 304 was always true

305 msg = f"Data leakage detected for '{feature}' during {stage}" 

306 else: 

307 msg = message 

308 

309 self.feature = feature 

310 self.stage = stage 

311 super().__init__(msg) 

312 

313 def __str__(self) -> str: 

314 return f"[DataLeakageError:{self.feature}] {self.message}"