This is an automated email from the ASF dual-hosted git repository.
shahar1 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 91a6a0fe8a8 Allow templated fields in AppFlow operators (#70440)
91a6a0fe8a8 is described below
commit 91a6a0fe8a832bc22be908eeac30eb5296979c9b
Author: Vincent Hsiao <[email protected]>
AuthorDate: Mon Jul 27 13:48:01 2026 +0800
Allow templated fields in AppFlow operators (#70440)
AppFlow operators validate template fields in constructors, so Jinja
expressions are rejected before Airflow can render them. Move those checks to
execution time so Dags can use templates while keeping the same runtime
validation.
---
.../providers/amazon/aws/operators/appflow.py | 63 +++++++++++++++-------
.../unit/amazon/aws/operators/test_appflow.py | 39 ++++++++++++++
.../ci/prek/validate_operators_init_exemptions.txt | 1 -
3 files changed, 84 insertions(+), 19 deletions(-)
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/operators/appflow.py
b/providers/amazon/src/airflow/providers/amazon/aws/operators/appflow.py
index 390a22679ea..9a60273b259 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/operators/appflow.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/appflow.py
@@ -84,8 +84,6 @@ class AppflowBaseOperator(AwsBaseOperator[AppflowHook]):
**kwargs,
) -> None:
super().__init__(**kwargs)
- if source is not None and source not in SUPPORTED_SOURCES:
- raise ValueError(f"{source} is not a supported source (options:
{SUPPORTED_SOURCES})!")
self.filter_date = filter_date
self.flow_name = flow_name
self.source = source
@@ -96,6 +94,8 @@ class AppflowBaseOperator(AwsBaseOperator[AppflowHook]):
self.wait_for_completion = wait_for_completion
def execute(self, context: Context) -> None:
+ self._validate_source()
+ self._validate_filter_date()
self.filter_date_parsed: datetime | None = (
datetime.fromisoformat(self.filter_date) if self.filter_date else
None
)
@@ -109,6 +109,13 @@ class AppflowBaseOperator(AwsBaseOperator[AppflowHook]):
self._run_flow(context)
+ def _validate_source(self) -> None:
+ if self.source is not None and self.source not in SUPPORTED_SOURCES:
+ raise ValueError(f"{self.source} is not a supported source
(options: {SUPPORTED_SOURCES})!")
+
+ def _validate_filter_date(self) -> None:
+ pass
+
def _get_connector_type(self) -> str:
response = self.hook.conn.describe_flow(flowName=self.flow_name)
connector_type = response["sourceFlowConfig"]["connectorType"]
@@ -190,8 +197,6 @@ class AppflowRunFullOperator(AppflowBaseOperator):
wait_for_completion: bool = True,
**kwargs,
) -> None:
- if source not in {"salesforce", "zendesk"}:
- raise ValueError(NOT_SUPPORTED_SOURCE_MSG.format(source=source,
entity="AppflowRunFullOperator"))
super().__init__(
source=source,
flow_name=flow_name,
@@ -203,6 +208,12 @@ class AppflowRunFullOperator(AppflowBaseOperator):
**kwargs,
)
+ def _validate_source(self) -> None:
+ if self.source not in {"salesforce", "zendesk"}:
+ raise ValueError(
+ NOT_SUPPORTED_SOURCE_MSG.format(source=self.source,
entity="AppflowRunFullOperator")
+ )
+
class AppflowRunBeforeOperator(AppflowBaseOperator):
"""
@@ -236,12 +247,6 @@ class AppflowRunBeforeOperator(AppflowBaseOperator):
wait_for_completion: bool = True,
**kwargs,
) -> None:
- if not filter_date:
- raise
ValueError(MANDATORY_FILTER_DATE_MSG.format(entity="AppflowRunBeforeOperator"))
- if source != "salesforce":
- raise ValueError(
- NOT_SUPPORTED_SOURCE_MSG.format(source=source,
entity="AppflowRunBeforeOperator")
- )
super().__init__(
source=source,
flow_name=flow_name,
@@ -253,6 +258,16 @@ class AppflowRunBeforeOperator(AppflowBaseOperator):
**kwargs,
)
+ def _validate_source(self) -> None:
+ if self.source != "salesforce":
+ raise ValueError(
+ NOT_SUPPORTED_SOURCE_MSG.format(source=self.source,
entity="AppflowRunBeforeOperator")
+ )
+
+ def _validate_filter_date(self) -> None:
+ if not self.filter_date:
+ raise
ValueError(MANDATORY_FILTER_DATE_MSG.format(entity="AppflowRunBeforeOperator"))
+
def _update_flow(self) -> None:
if not self.filter_date_parsed:
raise ValueError(f"Invalid filter_date argument parser value:
{self.filter_date_parsed}")
@@ -298,10 +313,6 @@ class AppflowRunAfterOperator(AppflowBaseOperator):
wait_for_completion: bool = True,
**kwargs,
) -> None:
- if not filter_date:
- raise
ValueError(MANDATORY_FILTER_DATE_MSG.format(entity="AppflowRunAfterOperator"))
- if source not in {"salesforce", "zendesk"}:
- raise ValueError(NOT_SUPPORTED_SOURCE_MSG.format(source=source,
entity="AppflowRunAfterOperator"))
super().__init__(
source=source,
flow_name=flow_name,
@@ -313,6 +324,16 @@ class AppflowRunAfterOperator(AppflowBaseOperator):
**kwargs,
)
+ def _validate_source(self) -> None:
+ if self.source not in {"salesforce", "zendesk"}:
+ raise ValueError(
+ NOT_SUPPORTED_SOURCE_MSG.format(source=self.source,
entity="AppflowRunAfterOperator")
+ )
+
+ def _validate_filter_date(self) -> None:
+ if not self.filter_date:
+ raise
ValueError(MANDATORY_FILTER_DATE_MSG.format(entity="AppflowRunAfterOperator"))
+
def _update_flow(self) -> None:
if not self.filter_date_parsed:
raise ValueError(f"Invalid filter_date argument parser value:
{self.filter_date_parsed}")
@@ -358,10 +379,6 @@ class AppflowRunDailyOperator(AppflowBaseOperator):
wait_for_completion: bool = True,
**kwargs,
) -> None:
- if not filter_date:
- raise
ValueError(MANDATORY_FILTER_DATE_MSG.format(entity="AppflowRunDailyOperator"))
- if source != "salesforce":
- raise ValueError(NOT_SUPPORTED_SOURCE_MSG.format(source=source,
entity="AppflowRunDailyOperator"))
super().__init__(
source=source,
flow_name=flow_name,
@@ -373,6 +390,16 @@ class AppflowRunDailyOperator(AppflowBaseOperator):
**kwargs,
)
+ def _validate_source(self) -> None:
+ if self.source != "salesforce":
+ raise ValueError(
+ NOT_SUPPORTED_SOURCE_MSG.format(source=self.source,
entity="AppflowRunDailyOperator")
+ )
+
+ def _validate_filter_date(self) -> None:
+ if not self.filter_date:
+ raise
ValueError(MANDATORY_FILTER_DATE_MSG.format(entity="AppflowRunDailyOperator"))
+
def _update_flow(self) -> None:
if not self.filter_date_parsed:
raise ValueError(f"Invalid filter_date argument parser value:
{self.filter_date_parsed}")
diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_appflow.py
b/providers/amazon/tests/unit/amazon/aws/operators/test_appflow.py
index 86f765808d5..961a70a9116 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_appflow.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_appflow.py
@@ -192,6 +192,45 @@ def test_run_daily(appflow_conn, ctx, waiter_mock):
)
[email protected]_test
+def test_run_daily_with_templated_validation_fields(appflow_conn, ctx,
waiter_mock):
+ operator = AppflowRunDailyOperator(
+ source="{{ params.source }}",
+ flow_name=FLOW_NAME,
+ source_field="{{ params.source_field }}",
+ filter_date="{{ params.filter_date }}",
+ poll_interval=0,
+ task_id=TASK_ID,
+ )
+ context = {
+ **ctx,
+ "params": {
+ "source": SOURCE,
+ "source_field": "col0",
+ "filter_date": "2022-05-26T00:00+00:00",
+ },
+ }
+
+ operator.render_template_fields(context)
+ operator.execute(context)
+
+ run_assertions_base(
+ appflow_conn,
+ [
+ {
+ "taskType": "Filter",
+ "connectorOperator": {"Salesforce": "BETWEEN"},
+ "sourceFields": ["col0"],
+ "taskProperties": {
+ "DATA_TYPE": "datetime",
+ "LOWER_BOUND": "1653523199999",
+ "UPPER_BOUND": "1653609600000",
+ },
+ }
+ ],
+ )
+
+
@pytest.mark.db_test
def test_short_circuit(appflow_conn, ctx):
with mock.patch("airflow.models.TaskInstance.xcom_pull") as mock_xcom_pull:
diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt
b/scripts/ci/prek/validate_operators_init_exemptions.txt
index e192ef70baf..47ca0504b84 100644
--- a/scripts/ci/prek/validate_operators_init_exemptions.txt
+++ b/scripts/ci/prek/validate_operators_init_exemptions.txt
@@ -6,7 +6,6 @@
# Fixing a class (moving template-field validation/transformation out of
__init__ into
# execute()) MUST remove its entry in the same PR — the hook fails on stale
entries.
# Burn-down tracked at https://github.com/apache/airflow/issues/70296
-providers/amazon/src/airflow/providers/amazon/aws/operators/appflow.py::AppflowBaseOperator
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py::EmrAddStepsOperator
providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py::NeptuneStartDbClusterOperator
providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py::NeptuneStopDbClusterOperator
