ashb commented on code in PR #54505:
URL: https://github.com/apache/airflow/pull/54505#discussion_r2355026190
##########
airflow-core/src/airflow/exceptions.py:
##########
@@ -46,26 +42,24 @@ def serialize(self):
return f"{cls.__module__}.{cls.__name__}", (str(self),), {}
-class AirflowBadRequest(AirflowException):
- """Raise when the application or server cannot handle the request."""
-
- status_code = HTTPStatus.BAD_REQUEST
+class TaskNotFound(AirflowException):
Review Comment:
I have a feeling, that everything in core like this could now extend from
Exception/RuntimeException, as we don't have many cased in core we do an
`except AirflwoException`, so we could change them and then _this_
AirflowException could be a lazy import/compat shim to
airflow.sdk.exceptions.AirflowException?
##########
providers/amazon/tests/unit/amazon/aws/operators/test_datasync.py:
##########
@@ -441,9 +443,11 @@ def test_init_fails(self, mock_get_conn):
with pytest.raises(AirflowException):
self.set_up_operator(source_location_uri=None)
with pytest.raises(AirflowException):
- self.set_up_operator(destination_location_uri=None)
+ self.set_up_operator(destination_location_uri=None,
task_id="datasync_task2")
Review Comment:
And here. Assuming this case is failing because of duplicate task id, this
doesn't belong in provider tests.
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -279,14 +279,14 @@ def _resolve_connection(self) -> dict[str, Any]:
if not self.spark_binary:
self.spark_binary = extra.get("spark-binary",
DEFAULT_SPARK_BINARY)
if self.spark_binary is not None and self.spark_binary not in
ALLOWED_SPARK_BINARIES:
- raise RuntimeError(
+ raise ValueError(
Review Comment:
This short of change should probably be done separately -- it's not a good
idea to mix provider runtime code changes along with ones that touch core and
sdk.
##########
providers/amazon/tests/unit/amazon/aws/operators/test_datasync.py:
##########
@@ -217,9 +217,11 @@ def test_init_fails(self, mock_get_conn):
with pytest.raises(AirflowException):
self.set_up_operator(source_location_uri=None)
with pytest.raises(AirflowException):
- self.set_up_operator(destination_location_uri=None)
+ self.set_up_operator(destination_location_uri=None,
task_id="datasync_task2")
Review Comment:
DuplicatedTaskIdException probably _should_ exist in core, no? It's
something dag authors might hit?
Also however: this test is pointless in a provider! It's testing behaviuor
of core functionality, nothing provider specific.
Lets open a speparate PR to remove this test first, as it doesn't belong in
providers
##########
airflow-core/tests/unit/core/test_core.py:
##########
@@ -24,10 +24,10 @@
import pytest
from airflow._shared.timezones.timezone import datetime
-from airflow.exceptions import AirflowTaskTimeout
from airflow.models.baseoperator import BaseOperator
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.python import PythonOperator
+from airflow.sdk.exceptions import AirflowTaskTimeout
Review Comment:
Why does a test in core depend on an exception from the SDK? Should the test
that uses it be moved to sdk?
##########
airflow-core/tests/unit/dags/test_assets.py:
##########
@@ -19,11 +19,11 @@
from datetime import datetime
-from airflow.exceptions import AirflowFailException, AirflowSkipException
from airflow.models.dag import DAG
Review Comment:
Unrelated to this PR nit: this should be changed too
##########
providers/amazon/src/airflow/providers/amazon/aws/hooks/athena_sql.py:
##########
@@ -119,7 +119,7 @@ def conn_config(self) -> AwsConnectionWrapper:
raise ValueError(
f"Encountered non-JSON in `extra` field for connection
{self.aws_conn_id!r}."
)
- except AirflowNotFoundException:
+ except (AirflowNotFoundException, RuntimeError):
Review Comment:
In this case do we need it -- won't it import AirflowNotFouncException from
the right place?
What error does a get_connection throw when the conn is not found?
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/version_compat.py:
##########
@@ -39,7 +39,13 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
from airflow.models.xcom import XCOM_RETURN_KEY
from airflow.sdk import BaseHook, BaseOperator
from airflow.sdk.definitions.context import context_merge
+ from airflow.sdk.exceptions import AirflowSkipException, TaskDeferred
Review Comment:
This wont be in 3.1, it'll be in 3.2. This is also why version-based checks
are generally not the right tool.
try/except ImportError is better here.
##########
airflow-core/src/airflow/models/dagbag.py:
##########
@@ -45,6 +45,7 @@
AirflowClusterPolicyError,
AirflowClusterPolicySkipDag,
AirflowClusterPolicyViolation,
+ AirflowDagCycleException, # type: ignore[no-redef]
Review Comment:
Won't this cause it to use the old/deprecated name and issue a warning?
i.e. shouldn't this _always_ be `airflow.sdk.exceptions import
AirflowDagCycleException` now?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
