Coverage for src/dataknobs_data/exceptions.py: 49%

1"""Custom exceptions for the dataknobs_data package.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7if TYPE_CHECKING: 

8 from dataknobs_data.query import Query 

9 

10 

11class DataknobsDataError(Exception): 

12 """Base exception for all dataknobs_data errors.""" 

13 

14 pass 

15 

16 

17class RecordNotFoundError(DataknobsDataError): 

18 """Raised when a requested record is not found.""" 

19 

20 def __init__(self, id: str): 

21 self.id = id 

22 super().__init__(f"Record with ID '{id}' not found") 

23 

24 

25class RecordValidationError(DataknobsDataError): 

26 """Raised when record validation fails.""" 

27 

28 def __init__(self, message: str, field_name: str | None = None): 

29 self.field_name = field_name 

30 if field_name: 

31 message = f"Field '{field_name}': {message}" 

32 super().__init__(message) 

33 

34 

35class FieldTypeError(DataknobsDataError): 

36 """Raised when a field type operation fails.""" 

37 

38 def __init__(self, field_name: str, expected_type: str, actual_type: str): 

39 self.field_name = field_name 

40 self.expected_type = expected_type 

41 self.actual_type = actual_type 

42 super().__init__( 

43 f"Field '{field_name}' type mismatch: expected {expected_type}, got {actual_type}" 

44 ) 

45 

46 

47class DatabaseError(DataknobsDataError): 

48 """General database error.""" 

49 

50 pass 

51 

52 

53class DatabaseConnectionError(DataknobsDataError): 

54 """Raised when database connection fails.""" 

55 

56 def __init__(self, backend: str, message: str): 

57 self.backend = backend 

58 super().__init__(f"Failed to connect to {backend} backend: {message}") 

59 

60 

61class DatabaseOperationError(DataknobsDataError): 

62 """Raised when a database operation fails.""" 

63 

64 def __init__(self, operation: str, message: str): 

65 self.operation = operation 

66 super().__init__(f"Database operation '{operation}' failed: {message}") 

67 

68 

69class QueryError(DataknobsDataError): 

70 """Raised when query execution fails.""" 

71 

72 def __init__(self, message: str, query: Query | None = None): 

73 self.query = query 

74 super().__init__(f"Query error: {message}") 

75 

76 

77class SerializationError(DataknobsDataError): 

78 """Raised when serialization/deserialization fails.""" 

79 

80 def __init__(self, format: str, message: str): 

81 self.format = format 

82 super().__init__(f"Serialization error ({format}): {message}") 

83 

84 

85class DataFormatError(DataknobsDataError): 

86 """Raised when data format is invalid or unsupported.""" 

87 

88 def __init__(self, format: str, message: str): 

89 self.format = format 

90 super().__init__(f"Data format error ({format}): {message}") 

91 

92 

93class BackendNotFoundError(DataknobsDataError): 

94 """Raised when a requested backend is not available.""" 

95 

96 def __init__(self, backend: str, available: list | None = None): 

97 self.backend = backend 

98 self.available = available or [] 

99 message = f"Backend '{backend}' not found" 

100 if self.available: 

101 message += f". Available backends: {', '.join(self.available)}" 

102 super().__init__(message) 

103 

104 

105class ConfigurationError(DataknobsDataError): 

106 """Raised when configuration is invalid.""" 

107 

108 def __init__(self, parameter: str, message: str): 

109 self.parameter = parameter 

110 super().__init__(f"Configuration error for '{parameter}': {message}") 

111 

112 

113class ConcurrencyError(DataknobsDataError): 

114 """Raised when a concurrency conflict occurs.""" 

115 

116 def __init__(self, message: str): 

117 super().__init__(f"Concurrency error: {message}") 

118 

119 

120class TransactionError(DataknobsDataError): 

121 """Raised when a transaction fails.""" 

122 

123 def __init__(self, message: str): 

124 super().__init__(f"Transaction error: {message}") 

125 

126 

127class MigrationError(DataknobsDataError): 

128 """Raised when data migration fails.""" 

129 

130 def __init__(self, source: str, target: str, message: str): 

131 self.source = source 

132 self.target = target 

133 super().__init__(f"Migration from {source} to {target} failed: {message}")