Lee-W commented on code in PR #50487:
URL: https://github.com/apache/airflow/pull/50487#discussion_r2084918140
##########
providers/standard/src/airflow/providers/standard/sensors/external_task.py:
##########
@@ -455,23 +461,25 @@ def _check_for_existence(self, session) -> None:
dag_to_wait = DagModel.get_current(self.external_dag_id, session)
if not dag_to_wait:
- raise AirflowException(f"The external DAG {self.external_dag_id}
does not exist.")
+ raise AirflowExternalTaskSensorException(
+ f"The external DAG {self.external_dag_id} does not exist."
+ )
if not os.path.exists(correct_maybe_zipped(dag_to_wait.fileloc)):
- raise AirflowException(f"The external DAG {self.external_dag_id}
was deleted.")
+ raise AirflowExternalTaskSensorException(f"The external DAG
{self.external_dag_id} was deleted.")
if self.external_task_ids:
refreshed_dag_info =
DagBag(dag_to_wait.fileloc).get_dag(self.external_dag_id)
for external_task_id in self.external_task_ids:
if not refreshed_dag_info.has_task(external_task_id):
- raise AirflowException(
+ raise AirflowExternalTaskSensorException(
f"The external task {external_task_id} in DAG
{self.external_dag_id} does not exist."
)
if self.external_task_group_id:
refreshed_dag_info =
DagBag(dag_to_wait.fileloc).get_dag(self.external_dag_id)
if not
refreshed_dag_info.has_task_group(self.external_task_group_id):
- raise AirflowException(
+ raise AirflowExternalTaskSensorException(
Review Comment:
The exception created here is not a description and does not help much.
Maybe something like `ExternalDagNotExistsError` would be better
##########
providers/standard/src/airflow/providers/standard/sensors/external_task.py:
##########
@@ -455,23 +461,25 @@ def _check_for_existence(self, session) -> None:
dag_to_wait = DagModel.get_current(self.external_dag_id, session)
if not dag_to_wait:
- raise AirflowException(f"The external DAG {self.external_dag_id}
does not exist.")
+ raise AirflowExternalTaskSensorException(
+ f"The external DAG {self.external_dag_id} does not exist."
+ )
if not os.path.exists(correct_maybe_zipped(dag_to_wait.fileloc)):
- raise AirflowException(f"The external DAG {self.external_dag_id}
was deleted.")
+ raise AirflowExternalTaskSensorException(f"The external DAG
{self.external_dag_id} was deleted.")
if self.external_task_ids:
refreshed_dag_info =
DagBag(dag_to_wait.fileloc).get_dag(self.external_dag_id)
for external_task_id in self.external_task_ids:
if not refreshed_dag_info.has_task(external_task_id):
- raise AirflowException(
+ raise AirflowExternalTaskSensorException(
f"The external task {external_task_id} in DAG
{self.external_dag_id} does not exist."
)
if self.external_task_group_id:
refreshed_dag_info =
DagBag(dag_to_wait.fileloc).get_dag(self.external_dag_id)
if not
refreshed_dag_info.has_task_group(self.external_task_group_id):
- raise AirflowException(
+ raise AirflowExternalTaskSensorException(
Review Comment:
and we can create multiple exceptions for different kinds of errors
##########
providers/standard/src/airflow/providers/standard/sensors/external_task.py:
##########
@@ -24,9 +23,12 @@
from typing import TYPE_CHECKING, Any, Callable, ClassVar
from airflow.configuration import conf
-from airflow.exceptions import AirflowException, AirflowSkipException
+from airflow.exceptions import (
+ AirflowSkipException,
+)
Review Comment:
```suggestion
from airflow.exceptions import AirflowSkipException
```
nit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
