You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by am...@apache.org on 2025/07/30 13:10:36 UTC
(airflow) branch main updated: Remove conditional definition of ClassBasedListener by version (#53915)
This is an automated email from the ASF dual-hosted git repository.
amoghdesai pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 51c547d1ef3 Remove conditional definition of ClassBasedListener by version (#53915)
51c547d1ef3 is described below
commit 51c547d1ef36fe8e44f19f40c2e07ad474075da1
Author: Amogh Desai <am...@gmail.com>
AuthorDate: Wed Jul 30 18:40:29 2025 +0530
Remove conditional definition of ClassBasedListener by version (#53915)
---
.../tests/unit/listeners/class_listener.py | 113 ++++++++-------------
1 file changed, 40 insertions(+), 73 deletions(-)
diff --git a/airflow-core/tests/unit/listeners/class_listener.py b/airflow-core/tests/unit/listeners/class_listener.py
index 49a018ea9d8..ec9b39cc316 100644
--- a/airflow-core/tests/unit/listeners/class_listener.py
+++ b/airflow-core/tests/unit/listeners/class_listener.py
@@ -20,80 +20,47 @@ from __future__ import annotations
from airflow.listeners import hookimpl
from airflow.utils.state import DagRunState, TaskInstanceState
-from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS
-if AIRFLOW_V_3_0_PLUS:
-
- class ClassBasedListener:
- def __init__(self):
- self.started_component = None
- self.stopped_component = None
- self.state = []
-
- @hookimpl
- def on_starting(self, component):
- self.started_component = component
- self.state.append(DagRunState.RUNNING)
-
- @hookimpl
- def before_stopping(self, component):
- global stopped_component
- stopped_component = component
- self.state.append(DagRunState.SUCCESS)
-
- @hookimpl
- def on_task_instance_running(self, previous_state, task_instance):
- self.state.append(TaskInstanceState.RUNNING)
-
- @hookimpl
- def on_task_instance_success(self, previous_state, task_instance):
- self.state.append(TaskInstanceState.SUCCESS)
-
- @hookimpl
- def on_task_instance_failed(self, previous_state, task_instance, error: None | str | BaseException):
- self.state.append(TaskInstanceState.FAILED)
-
- @hookimpl
- def on_dag_run_running(self, dag_run, msg: str):
- self.state.append(DagRunState.RUNNING)
-
- @hookimpl
- def on_dag_run_success(self, dag_run, msg: str):
- self.state.append(DagRunState.SUCCESS)
-
- @hookimpl
- def on_dag_run_failed(self, dag_run, msg: str):
- self.state.append(DagRunState.FAILED)
-else:
-
- class ClassBasedListener: # type: ignore[no-redef]
- def __init__(self):
- self.started_component = None
- self.stopped_component = None
- self.state = []
-
- @hookimpl
- def on_starting(self, component):
- self.started_component = component
- self.state.append(DagRunState.RUNNING)
-
- @hookimpl
- def before_stopping(self, component):
- global stopped_component
- stopped_component = component
- self.state.append(DagRunState.SUCCESS)
-
- @hookimpl
- def on_task_instance_running(self, previous_state, task_instance):
- self.state.append(TaskInstanceState.RUNNING)
-
- @hookimpl
- def on_task_instance_success(self, previous_state, task_instance):
- self.state.append(TaskInstanceState.SUCCESS)
-
- @hookimpl
- def on_task_instance_failed(self, previous_state, task_instance, error: None | str | BaseException):
- self.state.append(TaskInstanceState.FAILED)
+class ClassBasedListener:
+ def __init__(self):
+ self.started_component = None
+ self.stopped_component = None
+ self.state = []
+
+ @hookimpl
+ def on_starting(self, component):
+ self.started_component = component
+ self.state.append(DagRunState.RUNNING)
+
+ @hookimpl
+ def before_stopping(self, component):
+ global stopped_component
+ stopped_component = component
+ self.state.append(DagRunState.SUCCESS)
+
+ @hookimpl
+ def on_task_instance_running(self, previous_state, task_instance):
+ self.state.append(TaskInstanceState.RUNNING)
+
+ @hookimpl
+ def on_task_instance_success(self, previous_state, task_instance):
+ self.state.append(TaskInstanceState.SUCCESS)
+
+ @hookimpl
+ def on_task_instance_failed(self, previous_state, task_instance, error: None | str | BaseException):
+ self.state.append(TaskInstanceState.FAILED)
+
+ @hookimpl
+ def on_dag_run_running(self, dag_run, msg: str):
+ self.state.append(DagRunState.RUNNING)
+
+ @hookimpl
+ def on_dag_run_success(self, dag_run, msg: str):
+ self.state.append(DagRunState.SUCCESS)
+
+ @hookimpl
+ def on_dag_run_failed(self, dag_run, msg: str):
+ self.state.append(DagRunState.FAILED)
def clear():