You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by po...@apache.org on 2024/11/01 13:10:59 UTC
(airflow) branch test-standard-provider-with-pre-2-10 updated (477abdfd993 -> a9300110e32)
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a change to branch test-standard-provider-with-pre-2-10
in repository https://gitbox.apache.org/repos/asf/airflow.git
omit 477abdfd993 Test standard provider with Airflow 2.8 and 2.9
new a9300110e32 Test standard provider with Airflow 2.8 and 2.9
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (477abdfd993)
\
N -- N -- N refs/heads/test-standard-provider-with-pre-2-10 (a9300110e32)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../airflow/providers/standard/operators/python.py | 54 +++++++++++++++-------
.../providers/standard/sensors/date_time.py | 18 +++++++-
.../src/airflow/providers/standard/sensors/time.py | 18 +++++++-
3 files changed, 71 insertions(+), 19 deletions(-)
(airflow) 01/01: Test standard provider with Airflow 2.8 and 2.9
Posted by po...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch test-standard-provider-with-pre-2-10
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit a9300110e3240dcfdd486529c78a4066b91642f5
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Thu Oct 31 19:23:01 2024 +0100
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 | 4 +-
.../airflow/providers/standard/operators/python.py | 54 +++++++++++++++-------
.../providers/standard/sensors/date_time.py | 18 +++++++-
.../src/airflow/providers/standard/sensors/time.py | 18 +++++++-
4 files changed, 73 insertions(+), 21 deletions(-)
diff --git a/dev/breeze/src/airflow_breeze/global_constants.py b/dev/breeze/src/airflow_breeze/global_constants.py
index b309620cd47..036cedd900c 100644
--- a/dev/breeze/src/airflow_breeze/global_constants.py
+++ b/dev/breeze/src/airflow_breeze/global_constants.py
@@ -566,13 +566,13 @@ BASE_PROVIDERS_COMPATIBILITY_CHECKS: list[dict[str, str | list[str]]] = [
{
"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",
},
{
diff --git a/providers/src/airflow/providers/standard/operators/python.py b/providers/src/airflow/providers/standard/operators/python.py
index 4d908f6ba68..a8f3baeb1be 100644
--- a/providers/src/airflow/providers/standard/operators/python.py
+++ b/providers/src/airflow/providers/standard/operators/python.py
@@ -55,15 +55,18 @@ from airflow.providers.standard.utils.python_virtualenv import prepare_virtualen
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
+from tests_common.test_utils.compat import AIRFLOW_V_2_10_PLUS
+
log = logging.getLogger(__name__)
AIRFLOW_VERSION = Version(airflow_version)
+AIRFLOW_V_10_0_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:
@@ -199,7 +202,11 @@ class PythonOperator(BaseOperator):
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_10_0_PLUS:
+ from airflow.utils.context import context_get_outlet_events
+
+ self._asset_events = context_get_outlet_events(context)
return_value = self.execute_callable()
if self.show_return_value_in_logs:
@@ -218,7 +225,13 @@ class PythonOperator(BaseOperator):
: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)
@@ -563,18 +576,25 @@ class _BasePythonVirtualenvOperator(PythonOperator, metaclass=ABCMeta):
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.")
diff --git a/providers/src/airflow/providers/standard/sensors/date_time.py b/providers/src/airflow/providers/standard/sensors/date_time.py
index 20a6a484e05..b3da7a1d76b 100644
--- a/providers/src/airflow/providers/standard/sensors/date_time.py
+++ b/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
diff --git a/providers/src/airflow/providers/standard/sensors/time.py b/providers/src/airflow/providers/standard/sensors/time.py
index 6dba2628fce..4cd7e3c93e7 100644
--- a/providers/src/airflow/providers/standard/sensors/time.py
+++ b/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