Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/airflow/exceptions.py: 52%

1# 

2# Licensed to the Apache Software Foundation (ASF) under one 

3# or more contributor license agreements. See the NOTICE file 

4# distributed with this work for additional information 

5# regarding copyright ownership. The ASF licenses this file 

6# to you under the Apache License, Version 2.0 (the 

7# "License"); you may not use this file except in compliance 

8# with the License. You may obtain a copy of the License at 

9# 

10# http://www.apache.org/licenses/LICENSE-2.0 

11# 

12# Unless required by applicable law or agreed to in writing, 

13# software distributed under the License is distributed on an 

14# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 

15# KIND, either express or implied. See the License for the 

16# specific language governing permissions and limitations 

17# under the License. 

18# Note: Any AirflowException raised is expected to cause the TaskInstance 

19# to be marked in an ERROR state 

20"""Exceptions used by Airflow.""" 

21 

22from __future__ import annotations 

23 

24from http import HTTPStatus 

25from typing import TYPE_CHECKING, NamedTuple 

26 

27if TYPE_CHECKING: 

28 from airflow.models import DagRun 

29 

30# Re exporting AirflowConfigException from shared configuration 

31from airflow._shared.configuration.exceptions import AirflowConfigException as AirflowConfigException 

32 

33try: 

34 from airflow.sdk.exceptions import ( 

35 AirflowException, 

36 AirflowNotFoundException, 

37 AirflowOptionalProviderFeatureException as AirflowOptionalProviderFeatureException, 

38 AirflowRescheduleException as AirflowRescheduleException, 

39 AirflowTimetableInvalid as AirflowTimetableInvalid, 

40 NodeNotFound as NodeNotFound, 

41 ParamValidationError as ParamValidationError, 

42 TaskNotFound as TaskNotFound, 

43 ) 

44except ModuleNotFoundError: 

45 # When _AIRFLOW__AS_LIBRARY is set, airflow.sdk may not be installed. 

46 # In that case, we define fallback exception classes that mirror the SDK ones. 

47 class AirflowException(Exception): # type: ignore[no-redef] 

48 """Base exception for Airflow errors.""" 

49 

50 class AirflowNotFoundException(AirflowException): # type: ignore[no-redef] 

51 """Raise when a requested object is not found.""" 

52 

53 class AirflowTimetableInvalid(AirflowException): # type: ignore[no-redef] 

54 """Raise when a DAG has an invalid timetable.""" 

55 

56 class TaskNotFound(AirflowException): # type: ignore[no-redef] 

57 """Raise when a Task is not available in the system.""" 

58 

59 class NodeNotFound(TaskNotFound, KeyError): # type: ignore[no-redef] 

60 """Raise when attempting to access an invalid node (task or task group) using [] notation.""" 

61 

62 def __str__(self) -> str: 

63 return str(self.args[0]) if self.args else "" 

64 

65 class AirflowRescheduleException(AirflowException): # type: ignore[no-redef] 

66 """ 

67 Raise when the task should be re-scheduled at a later time. 

68 

69 :param reschedule_date: The date when the task should be rescheduled 

70 """ 

71 

72 def __init__(self, reschedule_date): 

73 super().__init__() 

74 self.reschedule_date = reschedule_date 

75 

76 def serialize(self): 

77 cls = self.__class__ 

78 return f"{cls.__module__}.{cls.__name__}", (), {"reschedule_date": self.reschedule_date} 

79 

80 class AirflowOptionalProviderFeatureException(AirflowException): # type: ignore[no-redef] 

81 """Raise by providers when imports are missing for optional provider features.""" 

82 

83 class ParamValidationError(AirflowException, ValueError): # type: ignore[no-redef] 

84 """Raise when DAG params fail validation.""" 

85 

86 

87class AirflowBadRequest(AirflowException): 

88 """Raise when the application or server cannot handle the request.""" 

89 

90 status_code = HTTPStatus.BAD_REQUEST 

91 

92 

93class InvalidStatsNameException(AirflowException): 

94 """Raise when name of the stats is invalid.""" 

95 

96 

97class AirflowInternalRuntimeError(BaseException): 

98 """ 

99 Airflow Internal runtime error. 

100 

101 Indicates that something really terrible happens during the Airflow execution. 

102 

103 :meta private: 

104 """ 

105 

106 

107class AirflowDagDuplicatedIdException(AirflowException): 

108 """Raise when a DAG's ID is already used by another DAG.""" 

109 

110 def __init__(self, dag_id: str, incoming: str, existing: str) -> None: 

111 super().__init__(dag_id, incoming, existing) 

112 self.dag_id = dag_id 

113 self.incoming = incoming 

114 self.existing = existing 

115 

116 def __str__(self) -> str: 

117 return f"Ignoring DAG {self.dag_id} from {self.incoming} - also found in {self.existing}" 

118 

119 

120class AirflowClusterPolicyViolation(AirflowException): 

121 """Raise when there is a violation of a Cluster Policy in DAG definition.""" 

122 

123 

124class AirflowClusterPolicySkipDag(AirflowException): 

125 """Raise when skipping dag is needed in Cluster Policy.""" 

126 

127 

128class AirflowClusterPolicyError(AirflowException): 

129 """Raise for a Cluster Policy other than AirflowClusterPolicyViolation or AirflowClusterPolicySkipDag.""" 

130 

131 

132class DagNotFound(AirflowNotFoundException): 

133 """Raise when a DAG is not available in the system.""" 

134 

135 

136class DagCodeNotFound(AirflowNotFoundException): 

137 """Raise when a DAG code is not available in the system.""" 

138 

139 

140class DagRunNotFound(AirflowNotFoundException): 

141 """Raise when a DAG Run is not available in the system.""" 

142 

143 

144class DagVersionNotFound(AirflowNotFoundException): 

145 """Raised when a DagVersion for the given dag_id / bundle_version is not found.""" 

146 

147 

148class DagNotPartitionedError(ValueError): 

149 """Raise when a partition_key is supplied for a Dag that is not partitioned.""" 

150 

151 

152class InvalidPartitionKeyError(ValueError): 

153 """ 

154 Raise when a partition_key value is invalid. 

155 

156 1. empty or exceeds the maximum allowed length 

157 2. cannot be decoded to a partition_date by the timetable 

158 """ 

159 

160 

161class DagRunAlreadyExists(AirflowBadRequest): 

162 """Raise when creating a DAG run for DAG which already has DAG run entry.""" 

163 

164 def __init__(self, dag_run: DagRun) -> None: 

165 super().__init__(f"A DAG Run already exists for DAG {dag_run.dag_id} with run id {dag_run.run_id}") 

166 self.dag_run = dag_run 

167 

168 def serialize(self): 

169 cls = self.__class__ 

170 # Note the DagRun object will be detached here and fails serialization, we need to create a new one 

171 from airflow.models import DagRun 

172 

173 dag_run = DagRun( 

174 state=self.dag_run.state, 

175 dag_id=self.dag_run.dag_id, 

176 run_id=self.dag_run.run_id, 

177 run_type=self.dag_run.run_type, 

178 ) 

179 dag_run.id = self.dag_run.id 

180 return ( 

181 f"{cls.__module__}.{cls.__name__}", 

182 (), 

183 {"dag_run": dag_run}, 

184 ) 

185 

186 

187class SerializationError(AirflowException): 

188 """A problem occurred when trying to serialize something.""" 

189 

190 

191class TaskInstanceNotFound(AirflowNotFoundException): 

192 """Raise when a task instance is not available in the system.""" 

193 

194 

195class NotMapped(Exception): 

196 """Raise if a task is neither mapped nor has any parent mapped groups.""" 

197 

198 

199class PoolNotFound(AirflowNotFoundException): 

200 """Raise when a Pool is not available in the system.""" 

201 

202 

203class FileSyntaxError(NamedTuple): 

204 """Information about a single error in a file.""" 

205 

206 line_no: int | None 

207 message: str 

208 

209 def __str__(self): 

210 return f"{self.message}. Line number: {str(self.line_no)}," 

211 

212 

213class AirflowFileParseException(AirflowException): 

214 """ 

215 Raises when connection or variable file can not be parsed. 

216 

217 :param msg: The human-readable description of the exception 

218 :param file_path: A processed file that contains errors 

219 :param parse_errors: File syntax errors 

220 """ 

221 

222 def __init__(self, msg: str, file_path: str, parse_errors: list[FileSyntaxError]) -> None: 

223 super().__init__(msg) 

224 self.msg = msg 

225 self.file_path = file_path 

226 self.parse_errors = parse_errors 

227 

228 def __str__(self): 

229 from airflow.utils.code_utils import prepare_code_snippet 

230 from airflow.utils.platform import is_tty 

231 

232 result = f"{self.msg}\nFilename: {self.file_path}\n\n" 

233 

234 for error_no, parse_error in enumerate(self.parse_errors, 1): 

235 result += "=" * 20 + f" Parse error {error_no:3} " + "=" * 20 + "\n" 

236 result += f"{parse_error.message}\n" 

237 if parse_error.line_no: 

238 result += f"Line number: {parse_error.line_no}\n" 

239 if parse_error.line_no and is_tty(): 

240 result += "\n" + prepare_code_snippet(self.file_path, parse_error.line_no) + "\n" 

241 

242 return result 

243 

244 

245class AirflowUnsupportedFileTypeException(AirflowException): 

246 """Raise when a file type is not supported.""" 

247 

248 

249class ConnectionNotUnique(AirflowException): 

250 """Raise when multiple values are found for the same connection ID.""" 

251 

252 

253class VariableNotUnique(AirflowException): 

254 """Raise when multiple values are found for the same variable name.""" 

255 

256 

257# The try/except handling is needed after we moved all k8s classes to cncf.kubernetes provider 

258# These two exceptions are used internally by Kubernetes Executor but also by PodGenerator, so we need 

259# to leave them here in case older version of cncf.kubernetes provider is used to run KubernetesPodOperator 

260# and it raises one of those exceptions. The code should be backwards compatible even if you import 

261# and try/except the exception using direct imports from airflow.exceptions. 

262# 1) if you have old provider, both provider and pod generator will throw the "airflow.exceptions" exception. 

263# 2) if you have new provider, both provider and pod generator will throw the 

264# "airflow.providers.cncf.kubernetes" as it will be imported here from the provider. 

265try: 

266 from airflow.providers.cncf.kubernetes.exceptions import PodMutationHookException 

267except ImportError: 

268 

269 class PodMutationHookException(AirflowException): # type: ignore[no-redef] 

270 """Raised when exception happens during Pod Mutation Hook execution.""" 

271 

272 

273try: 

274 from airflow.providers.cncf.kubernetes.exceptions import PodReconciliationError 

275except ImportError: 

276 

277 class PodReconciliationError(AirflowException): # type: ignore[no-redef] 

278 """Raised when an error is encountered while trying to merge pod configs.""" 

279 

280 

281class RemovedInAirflow4Warning(DeprecationWarning): 

282 """Issued for usage of deprecated features that will be removed in Airflow4.""" 

283 

284 deprecated_since: str | None = None 

285 "Indicates the airflow version that started raising this deprecation warning" 

286 

287 

288class AirflowProviderDeprecationWarning(DeprecationWarning): 

289 """Issued for usage of deprecated features of Airflow provider.""" 

290 

291 deprecated_provider_since: str | None = None 

292 "Indicates the provider version that started raising this deprecation warning" 

293 

294 

295class DeserializingResultError(ValueError): 

296 """Raised when an error is encountered while a pickling library deserializes a pickle file.""" 

297 

298 def __str__(self): 

299 return ( 

300 "Error deserializing result. Note that result deserialization " 

301 "is not supported across major Python versions. Cause: " + str(self.__cause__) 

302 ) 

303 

304 

305class UnknownExecutorException(ValueError): 

306 """Raised when an attempt is made to load an executor which is not configured.""" 

307 

308 

309class DeserializationError(Exception): 

310 """ 

311 Raised when a Dag cannot be deserialized. 

312 

313 This exception should be raised using exception chaining: 

314 `raise DeserializationError(dag_id) from original_exception` 

315 """ 

316 

317 def __init__(self, dag_id: str | None = None, message: str | None = None): 

318 self.dag_id = dag_id 

319 if message: 

320 # Use custom message if provided 

321 super().__init__(message) 

322 elif dag_id is None: 

323 super().__init__("Missing Dag ID in serialized Dag") 

324 else: 

325 super().__init__(f"An unexpected error occurred while trying to deserialize Dag '{dag_id}'") 

326 

327 

328class DagRunTypeNotAllowed(AirflowException): 

329 """Raised when a Dag does not allow the requested run type.""" 

330 

331 

332class AirflowClearRunningTaskException(AirflowException): 

333 """Raise when the user attempts to clear currently running tasks.""" 

334 

335 

336_DEPRECATED_EXCEPTIONS = { 

337 "AirflowDagCycleException", 

338 "AirflowFailException", 

339 "AirflowInactiveAssetInInletOrOutletException", 

340 "AirflowSensorTimeout", 

341 "AirflowSkipException", 

342 "AirflowTaskTerminated", 

343 "AirflowTaskTimeout", 

344 "DagRunTriggerException", 

345 "DownstreamTasksSkipped", 

346 "DuplicateTaskIdFound", 

347 "FailFastDagInvalidTriggerRule", 

348 "ParamValidationError", 

349 "TaskAlreadyInTaskGroup", 

350 "TaskDeferralError", 

351 "TaskDeferralTimeout", 

352 "TaskDeferred", 

353 "XComNotFound", 

354} 

355 

356 

357def __getattr__(name: str): 

358 """Provide backward compatibility for moved exceptions.""" 

359 if name in _DEPRECATED_EXCEPTIONS: 

360 import warnings 

361 

362 from airflow import DeprecatedImportWarning 

363 from airflow._shared.module_loading import import_string 

364 

365 target_path = f"airflow.sdk.exceptions.{name}" 

366 warnings.warn( 

367 f"airflow.exceptions.{name} is deprecated and will be removed in a future version. Use {target_path} instead.", 

368 DeprecatedImportWarning, 

369 stacklevel=2, 

370 ) 

371 return import_string(target_path) 

372 raise AttributeError(f"module '{__name__}' has no attribute '{name}'")