amoghrajesh commented on code in PR #53389:
URL: https://github.com/apache/airflow/pull/53389#discussion_r2284544536
##########
airflow-core/src/airflow/utils/__init__.py:
##########
@@ -45,6 +45,7 @@
"timeout": {
"timeout": "airflow.sdk.execution_time.timeout.timeout",
},
+ "trigger_rule": {"*": "airflow.task.trigger_rule"},
Review Comment:
Yeah sounds like a better place to move it to since its used at the task
level
##########
airflow-core/src/airflow/example_dags/example_latest_only_with_trigger.py:
##########
@@ -28,8 +28,7 @@
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.latest_only import LatestOnlyOperator
-from airflow.sdk import DAG
-from airflow.utils.trigger_rule import TriggerRule
+from airflow.sdk import DAG, TriggerRule
Review Comment:
Example Core dags look fine
##########
airflow-core/tests/unit/models/test_dag.py:
##########
@@ -1399,8 +1399,7 @@ def test_create_dagrun_job_id_is_set(self,
testing_dag_bundle):
def test_dag_add_task_checks_trigger_rule(self):
# A non fail stop dag should allow any trigger rule
- from airflow.exceptions import FailFastDagInvalidTriggerRule
- from airflow.utils.trigger_rule import TriggerRule
+ from airflow.sdk.exceptions import FailFastDagInvalidTriggerRule
Review Comment:
This isn't correct, we should not be importing from sdk in here.
##########
task-sdk/src/airflow/sdk/bases/operator.py:
##########
@@ -1111,9 +1110,11 @@ def __init__(
stacklevel=2,
)
- if not TriggerRule.is_valid(trigger_rule):
+ try:
+ TriggerRule(trigger_rule)
+ except ValueError:
raise ValueError(
- f"The trigger_rule must be one of
{TriggerRule.all_triggers()},"
+ f"The trigger_rule must be one of {[rule.value for rule in
TriggerRule]},"
Review Comment:
Looks nice.
##########
airflow-core/tests/unit/models/test_dag.py:
##########
@@ -1399,8 +1399,7 @@ def test_create_dagrun_job_id_is_set(self,
testing_dag_bundle):
def test_dag_add_task_checks_trigger_rule(self):
# A non fail stop dag should allow any trigger rule
- from airflow.exceptions import FailFastDagInvalidTriggerRule
- from airflow.utils.trigger_rule import TriggerRule
+ from airflow.sdk.exceptions import FailFastDagInvalidTriggerRule
Review Comment:
You could move this test to the task sdk where it belongs tbh.
##########
performance/src/performance_dags/performance_dag/performance_dag.py:
##########
@@ -57,7 +57,12 @@
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.sdk import chain
-from airflow.utils.trigger_rule import TriggerRule
+
+try:
+ from airflow.sdk import TriggerRule
+except ImportError:
+ # Compatibility for Airflow < 3.1
+ from airflow.utils.trigger_rule import TriggerRule # type:
ignore[no-redef,attr-defined]
Review Comment:
Performance dags are part of core if I am not mistaken. Not sure though.
##########
providers/amazon/tests/system/amazon/aws/example_appflow_run.py:
##########
@@ -40,7 +40,11 @@
from airflow.models.baseoperator import chain # type:
ignore[attr-defined,no-redef]
from airflow.models.dag import DAG # type:
ignore[attr-defined,no-redef,assignment]
-from airflow.utils.trigger_rule import TriggerRule
+try:
Review Comment:
All provider example dags look good too
##########
task-sdk/src/airflow/sdk/exceptions.py:
##########
@@ -68,3 +69,22 @@ def __str__(self) -> str:
for arg in self.args[1:]:
typename = f"{typename}[{type(arg).__qualname__}]"
return f"unmappable return type {typename!r}"
+
+
+class FailFastDagInvalidTriggerRule(AirflowException):
+ """Raise when a dag has 'fail_fast' enabled yet has a non-default trigger
rule."""
+
+ _allowed_rules = (TriggerRule.ALL_SUCCESS,
TriggerRule.ALL_DONE_SETUP_SUCCESS)
+
+ @classmethod
+ def check(cls, *, fail_fast: bool, trigger_rule: TriggerRule):
+ """
+ Check that fail_fast dag tasks have allowable trigger rules.
+
+ :meta private:
+ """
+ if fail_fast and trigger_rule not in cls._allowed_rules:
+ raise cls()
+
+ def __str__(self) -> str:
+ return f"A 'fail_fast' dag can only have {TriggerRule.ALL_SUCCESS}
trigger rule"
Review Comment:
As I understand this, its only raised and not caught in the sdk. If that's
the case, you should not import this in the core (even in tests), move those
tests to sdk.
##########
devel-common/src/tests_common/test_utils/compat.py:
##########
@@ -64,6 +64,13 @@
from airflow.utils.xcom import XCOM_RETURN_KEY # type:
ignore[no-redef,attr-defined]
+try:
+ from airflow.sdk import TriggerRule
+except ImportError:
+ # Compatibility for Airflow < 3.1
+ from airflow.utils.trigger_rule import TriggerRule # type:
ignore[no-redef,attr-defined]
Review Comment:
Since we are doing this, the tests could actually import from here, but that
can be a follow-up if you want.
Like this for ex: `from tests_common.test_utils.compat import
XCOM_RETURN_KEY`
--
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]
