Coverage for haystack/core/errors.py: 86%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5from typing import Any 

6 

7from haystack.dataclasses.breakpoints import Breakpoint, PipelineSnapshot 

8 

9 

10class PipelineError(Exception): 

11 pass 

12 

13 

14class PipelineRuntimeError(Exception): 

15 def __init__( 

16 self, 

17 component_name: str | None, 

18 component_type: type | None, 

19 message: str, 

20 pipeline_snapshot: PipelineSnapshot | None = None, 

21 *, 

22 pipeline_snapshot_file_path: str | None = None, 

23 ) -> None: 

24 self.component_name = component_name 

25 self.component_type = component_type 

26 self.pipeline_snapshot = pipeline_snapshot 

27 self.pipeline_snapshot_file_path = pipeline_snapshot_file_path 

28 super().__init__(message) 

29 

30 @classmethod 

31 def from_exception(cls, component_name: str, component_type: type, error: Exception) -> "PipelineRuntimeError": 

32 """ 

33 Create a PipelineRuntimeError from an exception. 

34 """ 

35 message = ( 

36 f"The following component failed to run:\n" 

37 f"Component name: '{component_name}'\n" 

38 f"Component type: '{component_type.__name__}'\n" 

39 f"Error: {str(error)}" 

40 ) 

41 return cls(component_name, component_type, message) 

42 

43 @classmethod 

44 def from_invalid_output(cls, component_name: str, component_type: type, output: Any) -> "PipelineRuntimeError": 

45 """ 

46 Create a PipelineRuntimeError from an invalid output. 

47 """ 

48 message = ( 

49 f"The following component returned an invalid output:\n" 

50 f"Component name: '{component_name}'\n" 

51 f"Component type: '{component_type.__name__}'\n" 

52 f"Expected a dictionary, but got {type(output).__name__} instead.\n" 

53 f"Check the component's output and ensure it is a valid dictionary." 

54 ) 

55 return cls(component_name, component_type, message) 

56 

57 

58class PipelineComponentsBlockedError(PipelineRuntimeError): 

59 def __init__(self) -> None: 

60 message = ( 

61 "Cannot run pipeline - all components are blocked. " 

62 "This typically happens when:\n" 

63 "1. There is no valid entry point for the pipeline\n" 

64 "2. There is a circular dependency preventing the pipeline from running\n" 

65 "Check the connections between these components and ensure all required inputs are provided." 

66 ) 

67 super().__init__(None, None, message) 

68 

69 

70class PipelineConnectError(PipelineError): 

71 pass 

72 

73 

74class PipelineValidationError(PipelineError): 

75 pass 

76 

77 

78class PipelineDrawingError(PipelineError): 

79 pass 

80 

81 

82class PipelineMaxComponentRuns(PipelineError): 

83 pass 

84 

85 

86class PipelineUnmarshalError(PipelineError): 

87 pass 

88 

89 

90class ComponentError(Exception): 

91 pass 

92 

93 

94class ComponentDeserializationError(Exception): 

95 pass 

96 

97 

98class DeserializationError(Exception): 

99 pass 

100 

101 

102class SerializationError(Exception): 

103 pass 

104 

105 

106class BreakpointException(Exception): 

107 """ 

108 Exception raised when a pipeline breakpoint is triggered. 

109 """ 

110 

111 def __init__( 

112 self, 

113 message: str, 

114 component: str | None = None, 

115 pipeline_snapshot: PipelineSnapshot | None = None, 

116 pipeline_snapshot_file_path: str | None = None, 

117 *, 

118 break_point: Breakpoint | None = None, 

119 ) -> None: 

120 super().__init__(message) 

121 self.component = component 

122 self.pipeline_snapshot = pipeline_snapshot 

123 self.pipeline_snapshot_file_path = pipeline_snapshot_file_path 

124 self._break_point = break_point 

125 

126 if self.pipeline_snapshot is None and self._break_point is None: 

127 raise ValueError("Either pipeline_snapshot or break_point must be provided.") 

128 

129 @classmethod 

130 def from_triggered_breakpoint(cls, break_point: Breakpoint) -> "BreakpointException": 

131 """ 

132 Create a BreakpointException from a triggered breakpoint. 

133 """ 

134 msg = f"Breaking at component {break_point.component_name} at visit count {break_point.visit_count}" 

135 return BreakpointException(message=msg, component=break_point.component_name, break_point=break_point) 

136 

137 @property 

138 def inputs(self) -> dict[str, Any] | None: 

139 """ 

140 Returns the current inputs of the pipeline at the breakpoint. 

141 """ 

142 if not self.pipeline_snapshot: 

143 return None 

144 return self.pipeline_snapshot.pipeline_state.inputs 

145 

146 @property 

147 def results(self) -> dict[str, Any] | None: 

148 """ 

149 Returns the current outputs of the pipeline at the breakpoint. 

150 """ 

151 if not self.pipeline_snapshot: 

152 return None 

153 return self.pipeline_snapshot.pipeline_state.pipeline_outputs 

154 

155 @property 

156 def break_point(self) -> Breakpoint: 

157 """ 

158 Returns the Breakpoint that caused this exception. 

159 

160 If a specific break point was provided during initialization, it is returned. 

161 Otherwise, if the pipeline snapshot contains a break point, that is returned. 

162 """ 

163 if self._break_point is not None: 

164 return self._break_point 

165 # Mypy doesn't know that pipeline_snapshot.break_point must not be None here based on the constructor check 

166 return self.pipeline_snapshot.break_point # type: ignore[union-attr] 

167 

168 

169class PipelineInvalidPipelineSnapshotError(Exception): 

170 """ 

171 Exception raised when a pipeline is resumed from an invalid snapshot. 

172 """ 

173 

174 def __init__(self, message: str) -> None: 

175 super().__init__(message)