Coverage for railway / core / errors.py: 55%

1"""Custom error types for Railway Framework.""" 

2 

3from typing import Any, Dict, Optional 

4 

5 

6class RailwayError(Exception): 

7 """Base exception for all Railway Framework errors.""" 

8 

9 def __init__( 

10 self, 

11 message: str, 

12 *, 

13 code: Optional[str] = None, 

14 hint: Optional[str] = None, 

15 retryable: bool = False, 

16 ): 

17 super().__init__(message) 

18 self.message = message 

19 self.code = code 

20 self.hint = hint 

21 self.retryable = retryable 

22 

23 def full_message(self) -> str: 

24 """Get full formatted error message.""" 

25 parts = [] 

26 

27 if self.code: 

28 parts.append(f"[{self.code}]") 

29 

30 parts.append(self.message) 

31 

32 if self.hint: 

33 parts.append(f"\nヒント: {self.hint}") 

34 

35 return " ".join(parts) 

36 

37 def to_dict(self) -> Dict[str, Any]: 

38 """Convert error to dictionary.""" 

39 return { 

40 "type": self.__class__.__name__, 

41 "message": self.message, 

42 "code": self.code, 

43 "hint": self.hint, 

44 "retryable": self.retryable, 

45 } 

46 

47 

48class ConfigurationError(RailwayError): 

49 """Error related to configuration issues.""" 

50 

51 def __init__( 

52 self, 

53 message: str, 

54 *, 

55 code: Optional[str] = None, 

56 hint: Optional[str] = None, 

57 config_key: Optional[str] = None, 

58 ): 

59 if hint is None: 59 ↛ 62line 59 didn't jump to line 62 because the condition on line 59 was always true

60 hint = "設定ファイル（config/*.yaml）または環境変数を確認してください。" 

61 

62 super().__init__(message, code=code, hint=hint, retryable=False) 

63 self.config_key = config_key 

64 

65 def to_dict(self) -> Dict[str, Any]: 

66 d = super().to_dict() 

67 d["config_key"] = self.config_key 

68 return d 

69 

70 

71class NodeError(RailwayError): 

72 """Error that occurred in a node.""" 

73 

74 def __init__( 

75 self, 

76 message: str, 

77 *, 

78 code: Optional[str] = None, 

79 hint: Optional[str] = None, 

80 retryable: bool = True, 

81 node_name: Optional[str] = None, 

82 original_error: Optional[Exception] = None, 

83 ): 

84 super().__init__(message, code=code, hint=hint, retryable=retryable) 

85 self.node_name = node_name 

86 self.original_error = original_error 

87 

88 def full_message(self) -> str: 

89 """Get full formatted error message with node info.""" 

90 parts = [] 

91 

92 if self.code: 92 ↛ 95line 92 didn't jump to line 95 because the condition on line 92 was always true

93 parts.append(f"[{self.code}]") 

94 

95 if self.node_name: 95 ↛ 98line 95 didn't jump to line 98 because the condition on line 95 was always true

96 parts.append(f"[{self.node_name}]") 

97 

98 parts.append(self.message) 

99 

100 if self.hint: 100 ↛ 103line 100 didn't jump to line 103 because the condition on line 100 was always true

101 parts.append(f"\nヒント: {self.hint}") 

102 

103 return " ".join(parts) 

104 

105 def to_dict(self) -> Dict[str, Any]: 

106 d = super().to_dict() 

107 d["node_name"] = self.node_name 

108 if self.original_error: 

109 d["original_error"] = { 

110 "type": type(self.original_error).__name__, 

111 "message": str(self.original_error), 

112 } 

113 return d 

114 

115 

116class PipelineError(RailwayError): 

117 """Error that occurred in a pipeline.""" 

118 

119 def __init__( 

120 self, 

121 message: str, 

122 *, 

123 code: Optional[str] = None, 

124 hint: Optional[str] = None, 

125 step_number: Optional[int] = None, 

126 step_name: Optional[str] = None, 

127 total_steps: Optional[int] = None, 

128 original_error: Optional[Exception] = None, 

129 ): 

130 super().__init__(message, code=code, hint=hint, retryable=False) 

131 self.step_number = step_number 

132 self.step_name = step_name 

133 self.total_steps = total_steps 

134 self.original_error = original_error 

135 

136 @property 

137 def remaining_steps(self) -> Optional[int]: 

138 """Get number of remaining steps after failure.""" 

139 if self.step_number is not None and self.total_steps is not None: 139 ↛ 141line 139 didn't jump to line 141 because the condition on line 139 was always true

140 return self.total_steps - self.step_number 

141 return None 

142 

143 def full_message(self) -> str: 

144 """Get full formatted error message with pipeline info.""" 

145 parts = [] 

146 

147 if self.code: 

148 parts.append(f"[{self.code}]") 

149 

150 if self.step_name and self.step_number: 

151 parts.append(f"Step {self.step_number} ({self.step_name}):") 

152 

153 parts.append(self.message) 

154 

155 if self.remaining_steps is not None and self.remaining_steps > 0: 

156 parts.append(f"(残り {self.remaining_steps} ステップはスキップされました)") 

157 

158 if self.hint: 

159 parts.append(f"\nヒント: {self.hint}") 

160 

161 return " ".join(parts) 

162 

163 def to_dict(self) -> Dict[str, Any]: 

164 d = super().to_dict() 

165 d["step_number"] = self.step_number 

166 d["step_name"] = self.step_name 

167 d["total_steps"] = self.total_steps 

168 d["remaining_steps"] = self.remaining_steps 

169 return d 

170 

171 

172class NetworkError(RailwayError): 

173 """Error related to network operations.""" 

174 

175 def __init__( 

176 self, 

177 message: str, 

178 *, 

179 code: Optional[str] = None, 

180 hint: Optional[str] = None, 

181 url: Optional[str] = None, 

182 status_code: Optional[int] = None, 

183 ): 

184 if hint is None: 184 ↛ 187line 184 didn't jump to line 187 because the condition on line 184 was always true

185 hint = "ネットワーク接続を確認してください。APIエンドポイントが正しいか確認してください。" 

186 

187 super().__init__(message, code=code, hint=hint, retryable=True) 

188 self.url = url 

189 self.status_code = status_code 

190 

191 def to_dict(self) -> Dict[str, Any]: 

192 d = super().to_dict() 

193 d["url"] = self.url 

194 d["status_code"] = self.status_code 

195 return d 

196 

197 

198class ValidationError(RailwayError): 

199 """Error related to data validation.""" 

200 

201 def __init__( 

202 self, 

203 message: str, 

204 *, 

205 code: Optional[str] = None, 

206 hint: Optional[str] = None, 

207 field: Optional[str] = None, 

208 value: Any = None, 

209 ): 

210 if hint is None: 210 ↛ 213line 210 didn't jump to line 213 because the condition on line 210 was always true

211 hint = "入力データの形式を確認してください。" 

212 

213 super().__init__(message, code=code, hint=hint, retryable=False) 

214 self.field = field 

215 self.value = value 

216 

217 def to_dict(self) -> Dict[str, Any]: 

218 d = super().to_dict() 

219 d["field"] = self.field 

220 d["value"] = repr(self.value) if self.value is not None else None 

221 return d 

222 

223 

224class RailwayTimeoutError(RailwayError): 

225 """Error when operation times out.""" 

226 

227 def __init__( 

228 self, 

229 message: str, 

230 *, 

231 code: Optional[str] = None, 

232 hint: Optional[str] = None, 

233 timeout_seconds: Optional[float] = None, 

234 ): 

235 if hint is None: 

236 hint = "タイムアウト値を増やすか、処理を分割してください。" 

237 

238 super().__init__(message, code=code, hint=hint, retryable=True) 

239 self.timeout_seconds = timeout_seconds 

240 

241 def to_dict(self) -> Dict[str, Any]: 

242 d = super().to_dict() 

243 d["timeout_seconds"] = self.timeout_seconds 

244 return d