Test standard provider with Airflow 2.8 and 2.9

The standard provider has now min version of Airflow = 2.8
since #43553, but we have not tested it for Airflow 2.8 and 2.9.

dev/breeze/src/airflow_breeze/global_constants.py

@@ -574,13 +574,13 @@ def get_airflow_extras():
     {
         "python-version": "3.9",
         "airflow-version": "2.8.4",
-        "remove-providers": "cloudant fab edge standard",
+        "remove-providers": "cloudant fab edge",
         "run-tests": "true",
     },
     {
         "python-version": "3.9",
         "airflow-version": "2.9.3",
-        "remove-providers": "cloudant edge standard",
+        "remove-providers": "cloudant edge",
         "run-tests": "true",
     },
     {

providers/src/airflow/providers/standard/operators/python.py

@@ -54,15 +54,16 @@
 from airflow.settings import _ENABLE_AIP_44
 from airflow.typing_compat import Literal
 from airflow.utils import hashlib_wrapper
-from airflow.utils.context import context_copy_partial, context_get_outlet_events, context_merge
+from airflow.utils.context import context_copy_partial, context_merge
 from airflow.utils.file import get_unique_dag_module_name
-from airflow.utils.operator_helpers import ExecutionCallableRunner, KeywordParameters
-from airflow.utils.process_utils import execute_in_subprocess
+from airflow.utils.operator_helpers import KeywordParameters
+from airflow.utils.process_utils import execute_in_subprocess, execute_in_subprocess_with_kwargs
 from airflow.utils.session import create_session
 
 log = logging.getLogger(__name__)
 
 AIRFLOW_VERSION = Version(airflow_version)
+AIRFLOW_V_2_10_PLUS = Version(AIRFLOW_VERSION.base_version) >= Version("2.10.0")
 AIRFLOW_V_3_0_PLUS = Version(AIRFLOW_VERSION.base_version) >= Version("3.0.0")
 
 if TYPE_CHECKING:
@@ -187,7 +188,15 @@ def __init__(
     def execute(self, context: Context) -> Any:
         context_merge(context, self.op_kwargs, templates_dict=self.templates_dict)
         self.op_kwargs = self.determine_kwargs(context)
-        self._asset_events = context_get_outlet_events(context)
+
+        if AIRFLOW_V_3_0_PLUS:
+            from airflow.utils.context import context_get_outlet_events
+
+            self._asset_events = context_get_outlet_events(context)
+        elif AIRFLOW_V_2_10_PLUS:
+            from airflow.utils.context import context_get_outlet_events
+
+            self._dataset_events = context_get_outlet_events(context)
 
         return_value = self.execute_callable()
         if self.show_return_value_in_logs:
@@ -206,7 +215,13 @@ def execute_callable(self) -> Any:
 
         :return: the return value of the call.
         """
-        runner = ExecutionCallableRunner(self.python_callable, self._asset_events, logger=self.log)
+        try:
+            from airflow.utils.operator_helpers import ExecutionCallableRunner
+
+            runner = ExecutionCallableRunner(self.python_callable, self._asset_events, logger=self.log)
+        except ImportError:
+            # Handle Pre Airflow 3.10 case where ExecutionCallableRunner was not available
+            return self.python_callable(*self.op_args, **self.op_kwargs)
         return runner.run(*self.op_args, **self.op_kwargs)
 
 
@@ -551,18 +566,25 @@ def _execute_python_callable_in_subprocess(self, python_path: Path):
                 env_vars.update(self.env_vars)
 
             try:
-                execute_in_subprocess(
-                    cmd=[
-                        os.fspath(python_path),
-                        os.fspath(script_path),
-                        os.fspath(input_path),
-                        os.fspath(output_path),
-                        os.fspath(string_args_path),
-                        os.fspath(termination_log_path),
-                        os.fspath(airflow_context_path),
-                    ],
-                    env=env_vars,
-                )
+                cmd: list[str] = [
+                    os.fspath(python_path),
+                    os.fspath(script_path),
+                    os.fspath(input_path),
+                    os.fspath(output_path),
+                    os.fspath(string_args_path),
+                    os.fspath(termination_log_path),
+                    os.fspath(airflow_context_path),
+                ]
+                if AIRFLOW_V_2_10_PLUS:
+                    execute_in_subprocess(
+                        cmd=cmd,
+                        env=env_vars,
+                    )
+                else:
+                    execute_in_subprocess_with_kwargs(
+                        cmd=cmd,
+                        env=env_vars,
+                    )
             except subprocess.CalledProcessError as e:
                 if e.returncode in self.skip_on_exit_code:
                     raise AirflowSkipException(f"Process exited with code {e.returncode}. Skipping.")

providers/src/airflow/providers/standard/sensors/date_time.py

@@ -18,10 +18,26 @@
 from __future__ import annotations
 
 import datetime
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, NoReturn, Sequence
 
 from airflow.sensors.base import BaseSensorOperator
-from airflow.triggers.base import StartTriggerArgs
+
+try:
+    from airflow.triggers.base import StartTriggerArgs
+except ImportError:
+    # TODO: Remove this when min airflow version is 2.10.0 for standard provider
+    @dataclass
+    class StartTriggerArgs:  # type: ignore[no-redef]
+        """Arguments required for start task execution from triggerer."""
+
+        trigger_cls: str
+        next_method: str
+        trigger_kwargs: dict[str, Any] | None = None
+        next_kwargs: dict[str, Any] | None = None
+        timeout: datetime.timedelta | None = None
+
+
 from airflow.triggers.temporal import DateTimeTrigger
 from airflow.utils import timezone
 

providers/src/airflow/providers/standard/sensors/time.py

@@ -18,10 +18,26 @@
 from __future__ import annotations
 
 import datetime
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, NoReturn
 
 from airflow.sensors.base import BaseSensorOperator
-from airflow.triggers.base import StartTriggerArgs
+
+try:
+    from airflow.triggers.base import StartTriggerArgs
+except ImportError:
+    # TODO: Remove this when min airflow version is 2.10.0 for standard provider
+    @dataclass
+    class StartTriggerArgs:  # type: ignore[no-redef]
+        """Arguments required for start task execution from triggerer."""
+
+        trigger_cls: str
+        next_method: str
+        trigger_kwargs: dict[str, Any] | None = None
+        next_kwargs: dict[str, Any] | None = None
+        timeout: datetime.timedelta | None = None
+
+
 from airflow.triggers.temporal import DateTimeTrigger
 from airflow.utils import timezone