Coverage for dataexcept/datascience_exceptions/operations.py: 78%

1"""Operational and deployment errors for ML systems.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any, Optional 

6 

7from .._validation import is_number 

8from ..redaction import redact_if_url 

9from .base import DataScienceError 

10 

11 

12class ModelSerializationError(DataScienceError): 

13 """ 

14 Raised when saving or loading a model fails. 

15 

16 Attributes: 

17 path: file path involved. 

18 original: underlying exception. 

19 """ 

20 

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

22 if not isinstance(path, str): 22 ↛ 23line 22 didn't jump to line 23 because the condition on line 22 was never true

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

24 if not isinstance(original, Exception): 24 ↛ 25line 24 didn't jump to line 25 because the condition on line 24 was never true

25 raise TypeError( 

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

27 ) 

28 

29 message = f"Failed to serialize to {path!r}: {original}" 

30 self.path = redact_if_url(path) 

31 self.original = original 

32 super().__init__(message) 

33 

34 def __str__(self) -> str: 

35 return f"[ModelSerializationError:{self.path}] {self.message}" 

36 

37 

38class DeploymentError(DataScienceError): 

39 """ 

40 Raised when deploying a model or pipeline fails. 

41 

42 Attributes: 

43 target: deployment target identifier. 

44 cause: optional detail. 

45 """ 

46 

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

48 if not isinstance(target, str): 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true

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

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

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

52 

53 msg = f"Deployment failed to '{target}'" 

54 if cause: 

55 msg += f": {cause}" 

56 

57 self.target = target 

58 self.cause = cause 

59 super().__init__(msg) 

60 

61 def __str__(self) -> str: 

62 return f"[DeploymentError:{self.target}] {self.message}" 

63 

64 

65class DataDriftError(DataScienceError): 

66 """ 

67 Raised when data drift is detected beyond threshold. 

68 

69 Attributes: 

70 feature: feature name. 

71 drift_score: computed drift metric. 

72 """ 

73 

74 def __init__( 

75 self, feature: str, drift_score: float, message: Optional[str] = None 

76 ) -> None: 

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

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

79 if not is_number(drift_score): 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true

80 raise TypeError( 

81 f"drift_score must be number, got {type(drift_score).__name__}" 

82 ) 

83 

84 self.feature = feature 

85 self.drift_score = float(drift_score) 

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

87 message = f"Data drift detected on '{feature}', score={drift_score:.4f}" 

88 

89 super().__init__(message) 

90 

91 def __str__(self) -> str: 

92 return f"[DataDriftError:{self.feature}] {self.message}" 

93 

94 

95class ResourceLimitError(DataScienceError): 

96 """ 

97 Raised when computation exceeds resources (memory, CPU). 

98 

99 Attributes: 

100 resource: 'memory', 'cpu', etc. 

101 limit: threshold exceeded. 

102 """ 

103 

104 def __init__(self, resource: str, limit: Any) -> None: 

105 if not isinstance(resource, str): 105 ↛ 106line 105 didn't jump to line 106 because the condition on line 105 was never true

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

107 

108 message = f"Resource limit exceeded: {resource} at {limit!r}" 

109 self.resource = resource 

110 self.limit = limit 

111 super().__init__(message) 

112 

113 def __str__(self) -> str: 

114 return f"[ResourceLimitError:{self.resource}] {self.message}" 

115 

116 

117class DataExportError(DataScienceError): 

118 """Failed to export or write data to destination.""" 

119 

120 def __init__(self, destination: str, original: Exception) -> None: 

121 if not isinstance(destination, str): 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true

122 raise TypeError( 

123 f"destination must be str, got {type(destination).__name__}" 

124 ) 

125 if not isinstance(original, Exception): 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true

126 raise TypeError( 

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

128 ) 

129 msg = f"Unable to export data to {destination}: {original}" 

130 self.destination = destination 

131 self.original = original 

132 super().__init__(msg)