Coverage for dataexcept/dataengineering_exceptions.py: 100%

1"""Custom exceptions for data engineering workflows.""" 

2 

3from __future__ import annotations 

4 

5from typing import Optional 

6 

7from .base import DataExceptError 

8from .redaction import redact_if_url 

9 

10 

11class DataEngineeringError(DataExceptError): 

12 """Base exception for data engineering errors.""" 

13 

14 pass 

15 

16 

17class ETLJobError(DataEngineeringError): 

18 """Raised when an ETL job fails to complete successfully.""" 

19 

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

21 """Initialize ETLJobError. 

22 

23 Args: 

24 job_name: Name of the ETL job. 

25 message: Optional custom error message. 

26 """ 

27 self.job_name = job_name 

28 default = f"ETL job '{job_name}' failed" 

29 super().__init__(message or default) 

30 

31 

32class SchemaEvolutionError(DataEngineeringError): 

33 """Raised when database schema evolution fails.""" 

34 

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

36 """Initialize SchemaEvolutionError. 

37 

38 Args: 

39 schema_version: Version of the schema being applied. 

40 reason: Optional explanation of the failure. 

41 """ 

42 self.schema_version = schema_version 

43 self.reason = reason 

44 msg = f"Schema evolution to {schema_version} failed" 

45 if reason: 

46 msg += f": {reason}" 

47 super().__init__(msg) 

48 

49 

50class DataTransformationError(DataEngineeringError): 

51 """Raised when a data transformation step fails.""" 

52 

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

54 """Initialize DataTransformationError. 

55 

56 Args: 

57 step: Name of the transformation step. 

58 details: Optional details about the failure. 

59 """ 

60 self.step = step 

61 self.details = details 

62 msg = f"Data transformation '{step}' failed" 

63 if details: 

64 msg += f": {details}" 

65 super().__init__(msg) 

66 

67 

68class BatchProcessingError(DataEngineeringError): 

69 """Raised when processing a data batch fails.""" 

70 

71 def __init__(self, batch_id: str, original: Optional[Exception] = None) -> None: 

72 """Initialize BatchProcessingError. 

73 

74 Args: 

75 batch_id: Identifier of the batch being processed. 

76 original: Optional underlying exception. 

77 """ 

78 self.batch_id = batch_id 

79 self.original = original 

80 msg = f"Batch '{batch_id}' processing failed" 

81 if original: 

82 msg += f": {original}" 

83 super().__init__(msg) 

84 

85 

86class DataWarehouseConnectionError(DataEngineeringError): 

87 """Raised when a connection to a data warehouse cannot be established.""" 

88 

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

90 """Initialize DataWarehouseConnectionError. 

91 

92 Args: 

93 warehouse: Identifier of the data warehouse. 

94 message: Optional custom error message. 

95 """ 

96 self.warehouse = warehouse 

97 default = f"Failed to connect to warehouse '{warehouse}'" 

98 super().__init__(message or default) 

99 

100 

101class MissingPartitionError(DataEngineeringError): 

102 """Raised when a required data partition is missing.""" 

103 

104 def __init__( 

105 self, partition: str, location: str, message: Optional[str] = None 

106 ) -> None: 

107 """Initialize MissingPartitionError. 

108 

109 Args: 

110 partition: Name of the missing partition. 

111 location: Data location checked for the partition. 

112 message: Optional custom error message. 

113 """ 

114 self.partition = partition 

115 self.location = redact_if_url(location) 

116 default = f"Partition '{partition}' not found at {location}" 

117 super().__init__(message or default) 

118 

119 

120__all__ = [ 

121 "DataEngineeringError", 

122 "ETLJobError", 

123 "SchemaEvolutionError", 

124 "DataTransformationError", 

125 "BatchProcessingError", 

126 "DataWarehouseConnectionError", 

127 "MissingPartitionError", 

128]