Coverage for dataexcept/pandas_exceptions.py: 81%

1"""Custom exceptions for pandas DataFrame operations.""" 

2 

3from __future__ import annotations 

4 

5from typing import Optional, Sequence 

6 

7from .base import DataExceptError 

8from .redaction import redact_if_url 

9 

10 

11class PandasError(DataExceptError): 

12 """Base exception for pandas-related errors.""" 

13 

14 

15class MissingColumnError(PandasError): 

16 """Raised when a required DataFrame column is missing. 

17 

18 Args: 

19 column: Name of the missing column. 

20 dataframe: Optional name of the DataFrame being inspected. 

21 """ 

22 

23 def __init__(self, column: str, dataframe: Optional[str] = None) -> None: 

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

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

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

27 raise TypeError( 

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

29 ) 

30 

31 self.column = column 

32 self.dataframe = dataframe 

33 name = f" in DataFrame '{dataframe}'" if dataframe else "" 

34 msg = f"Missing required column '{column}'{name}" 

35 super().__init__(msg) 

36 

37 def __str__(self) -> str: 

38 return f"[MissingColumnError] {self.args[0]}" 

39 

40 

41class DtypeMismatchError(PandasError): 

42 """Raised when a column has an unexpected dtype. 

43 

44 Args: 

45 column: Name of the column. 

46 expected: Sequence of allowed dtypes. 

47 found: Detected dtype for the column. 

48 """ 

49 

50 def __init__(self, column: str, expected: Sequence[str], found: str) -> None: 

51 if not isinstance(column, str): 51 ↛ 52line 51 didn't jump to line 52 because the condition on line 51 was never true

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

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

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

55 if not isinstance(expected, Sequence) or isinstance(expected, str): 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true

56 raise TypeError("expected must be a sequence of strings") 

57 if not all(isinstance(dt, str) for dt in expected): 57 ↛ 58line 57 didn't jump to line 58 because the condition on line 57 was never true

58 raise TypeError("expected must contain strings") 

59 

60 self.column = column 

61 self.expected = list(expected) 

62 self.found = found 

63 expected_fmt = ", ".join(self.expected) 

64 msg = f"Column '{column}' has dtype {found}; expected {expected_fmt}" 

65 super().__init__(msg) 

66 

67 def __str__(self) -> str: 

68 return f"[DtypeMismatchError:{self.column}] {self.args[0]}" 

69 

70 

71class IndexAlignmentError(PandasError): 

72 """Raised when DataFrame indices are misaligned for an operation. 

73 

74 Args: 

75 details: Optional details about the misalignment. 

76 """ 

77 

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

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

80 raise TypeError( 

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

82 ) 

83 msg = "DataFrame indices are misaligned" 

84 if details: 

85 msg += f": {details}" 

86 self.details = details 

87 super().__init__(msg) 

88 

89 def __str__(self) -> str: 

90 return f"[IndexAlignmentError] {self.args[0]}" 

91 

92 

93class MergeKeyError(PandasError): 

94 """Raised when merging DataFrames fails due to key issues. 

95 

96 Args: 

97 left_keys: Keys from the left DataFrame. 

98 right_keys: Keys from the right DataFrame. 

99 """ 

100 

101 def __init__(self, left_keys: Sequence[str], right_keys: Sequence[str]) -> None: 

102 # A bare string is a sequence of strings, so "id" would silently become 

103 # ['i', 'd']. Reject it, as DtypeMismatchError already does. 

104 for name, keys in (("left_keys", left_keys), ("right_keys", right_keys)): 

105 if isinstance(keys, str) or not all(isinstance(k, str) for k in keys): 

106 raise TypeError(f"{name} must be a sequence of strings, not a string") 

107 self.left_keys = list(left_keys) 

108 self.right_keys = list(right_keys) 

109 msg = f"Failed to merge on keys {self.left_keys} and {self.right_keys}" 

110 super().__init__(msg) 

111 

112 def __str__(self) -> str: 

113 return f"[MergeKeyError] {self.args[0]}" 

114 

115 

116class PandasIOError(PandasError): 

117 """Raised when reading from or writing to disk with pandas fails. 

118 

119 Args: 

120 path: File path involved in the operation. 

121 original: The underlying exception that was raised. 

122 """ 

123 

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

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

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

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

128 raise TypeError( 

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

130 ) 

131 self.path = redact_if_url(path) 

132 self.original = original 

133 msg = f"Pandas I/O operation failed on {path!r}: {original}" 

134 super().__init__(msg) 

135 

136 def __str__(self) -> str: 

137 return f"[PandasIOError] {self.args[0]}" 

138 

139 

140__all__ = [ 

141 "PandasError", 

142 "MissingColumnError", 

143 "DtypeMismatchError", 

144 "IndexAlignmentError", 

145 "MergeKeyError", 

146 "PandasIOError", 

147]