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 b30cd9b61be Allow template-field copies into StartTriggerArgs in
operator __init__ (#73040)
b30cd9b61be is described below
commit b30cd9b61be7f39a85b0a3e120ee8cc48617a593
Author: Shahar Epstein <[email protected]>
AuthorDate: Tue Sep 22 17:19:07 2026 +0300
Allow template-field copies into StartTriggerArgs in operator __init__
(#73040)
---
airflow-core/docs/howto/custom-operator.rst | 27 ++++
contributing-docs/05_pull_requests.rst | 25 ++++
scripts/ci/prek/validate_operators_init.py | 84 ++++++++++++-
.../ci/prek/validate_operators_init_exemptions.txt | 1 -
.../tests/ci/prek/test_validate_operators_init.py | 136 +++++++++++++++++++++
5 files changed, 270 insertions(+), 3 deletions(-)
diff --git a/airflow-core/docs/howto/custom-operator.rst
b/airflow-core/docs/howto/custom-operator.rst
index d12892355b1..9c65e9efbf5 100644
--- a/airflow-core/docs/howto/custom-operator.rst
+++ b/airflow-core/docs/howto/custom-operator.rst
@@ -354,6 +354,33 @@ still belongs in ``execute()``:
self.foo = foo
self.bar = bar
+5. Operators that support ``start_from_trigger`` may copy a templated field
verbatim into
+``start_trigger_args.trigger_kwargs`` under the field's own name. The
scheduler sends the task straight to
+the triggerer, which renders those entries itself, and ``execute()`` never
runs. The key has to match the
+field name and be an attribute of the trigger; nothing else in
``StartTriggerArgs`` is rendered, so a
+transformed value, or a key that is not a template field, is still invalid.
Storing one field's value
+under another template field's key is rendered, under that key, but is invalid
too — the entry no
+longer holds the field its key names:
+
+.. code-block:: python
+
+ class HelloOperator(BaseOperator):
+ template_fields = ("foo",)
+ start_trigger_args = StartTriggerArgs(
+ trigger_cls="my_package.triggers.HelloTrigger",
+ trigger_kwargs={},
+ next_method="execute_complete",
+ )
+
+ def __init__(self, foo, start_from_trigger=False) -> None:
+ self.foo = foo
+ self.start_from_trigger = start_from_trigger
+ if start_from_trigger:
+ self.start_trigger_args = dataclasses.replace(
+ self.start_trigger_args,
+ trigger_kwargs={"foo": self.foo}, # allowed: verbatim
copy under the field's name
+ )
+
When an operator inherits from a base operator and does not have a constructor
defined on its own, the limitations above
do not apply. However, the templated fields must be set properly in the parent
according to those limitations.
diff --git a/contributing-docs/05_pull_requests.rst
b/contributing-docs/05_pull_requests.rst
index b59d6bc362b..f8889d4eb13 100644
--- a/contributing-docs/05_pull_requests.rst
+++ b/contributing-docs/05_pull_requests.rst
@@ -464,6 +464,31 @@ Converting an existing truthiness check is not neutral: a
guard that raises when
passing, and that half is a value check. ``not exactly_one(a, b)`` carries
both, since it also
requires at least one.
+Operators that support ``start_from_trigger`` are the third exception. The
scheduler sends the task
+straight to the triggerer and ``execute`` never runs; the triggerer renders
the ``trigger_kwargs``
+entries whose key is both an operator template field and an attribute of the
trigger. Copying such
+a field verbatim into ``start_trigger_args`` under its own name is therefore
fine in the
+constructor:
+
+.. code-block:: python
+
+ def __init__(self, *, job: dict, start_from_trigger: bool = False,
**kwargs):
+ super().__init__(**kwargs)
+ self.job = job
+ self.start_from_trigger = start_from_trigger
+ if start_from_trigger:
+ self.start_trigger_args = dataclasses.replace(
+ self.start_trigger_args, trigger_kwargs={"job": self.job}
+ )
+
+A transformed value, a copy under a key that is not a template field of the
operator, and a template
+field passed as ``timeout`` or ``next_kwargs`` are never rendered and are
still flagged. A copy under
+the name of a *different* template field is rendered, under that key, but is
flagged as well — the
+entry no longer holds the field its key names.
+
+The hook only verifies the operator half of that pair: it cannot tell whether
the key is also an
+attribute of the trigger class, so a passing hook is not confirmation that the
kwarg will be rendered.
+
The reason for doing it is that we are working on a cleaning up our code to
have
`prek hook <../scripts/ci/prek/validate_operators_init.py>`_
that will make sure all the cases where logic (such as validation and complex
conversion)
diff --git a/scripts/ci/prek/validate_operators_init.py
b/scripts/ci/prek/validate_operators_init.py
index 86d25180508..2a2e85f2a0d 100755
--- a/scripts/ci/prek/validate_operators_init.py
+++ b/scripts/ci/prek/validate_operators_init.py
@@ -303,8 +303,8 @@ def _collect_sanctioned_uses(ctor: ast.FunctionDef,
template_fields: list[str])
``self.field = field``, ``self.field = field or <default>``, the equivalent
value-preserving ternaries, the local rebind ``field = field or
<default>``,
tuple assignments pairing names one-to-one, forwarding via
- ``super().__init__(field=field)``, and ``field is None`` / ``field is not
None``
- provision checks.
+ ``super().__init__(field=field)``, ``field is None`` / ``field is not
None``
+ provision checks, and verbatim copies into ``start_trigger_args``.
:param ctor: The constructor function node.
:param template_fields: The template fields of the class.
@@ -335,6 +335,14 @@ def _collect_sanctioned_uses(ctor: ast.FunctionDef,
template_fields: list[str])
name = _target_name(target)
if name is not None and name in template_fields:
mark(value, name)
+ elif isinstance(value, ast.Call) and
_is_start_trigger_args_assignment(target, value):
+ # The triggerer renders only the trigger_kwargs entries
whose key is both an
+ # operator template field and a trigger attribute
(airflow/triggers/base.py),
+ # so only a verbatim copy under the field's own name is
safe un-rendered.
+ # trigger_cls is a string, so the trigger-attribute half
is the author's to keep.
+ for key, item in _iter_trigger_kwargs_items(value):
+ if _target_name(item) == key:
+ sanctioned.add(id(item))
elif isinstance(node, ast.Call) and _is_super_init_call(node):
for keyword in node.keywords:
if keyword.arg is not None and keyword.arg in template_fields:
@@ -345,6 +353,78 @@ def _collect_sanctioned_uses(ctor: ast.FunctionDef,
template_fields: list[str])
return sanctioned
+def _is_call_to(func: ast.expr, name: str, *, modules: tuple[str, ...] = ())
-> bool:
+ """
+ Check whether a call target is the bare name ``name`` or
``<module>.<name>`` for a listed module.
+
+ Any other qualifier (``helper.replace``, ``factory.StartTriggerArgs``) is
rejected so an
+ unrelated callable cannot borrow a sanctioned name.
+
+ :param func: The ``func`` node of the call.
+ :param name: The callable name to match.
+ :param modules: Module names allowed as the attribute qualifier.
+ :return: True if the call target matches.
+ """
+ if isinstance(func, ast.Name):
+ return func.id == name
+ return (
+ isinstance(func, ast.Attribute)
+ and func.attr == name
+ and isinstance(func.value, ast.Name)
+ and func.value.id in modules
+ )
+
+
+def _is_start_trigger_args_assignment(target: ast.expr, value: ast.Call) ->
bool:
+ """
+ Check whether an assignment rebuilds ``self.start_trigger_args``.
+
+ Matches ``self.start_trigger_args = StartTriggerArgs(...)`` and
+ ``self.start_trigger_args = dataclasses.replace(self.start_trigger_args,
...)`` (also a bare
+ ``replace`` or ``copy.replace``). Aliased imports and other module
qualifiers are deliberately
+ not matched.
+
+ :param target: The assignment target.
+ :param value: The assigned call.
+ :return: True if the assignment constructs or copies ``StartTriggerArgs``.
+ """
+ if not (isinstance(target, ast.Attribute) and _target_name(target) ==
"start_trigger_args"):
+ return False
+ if _is_call_to(value.func, "StartTriggerArgs"):
+ return True
+ return (
+ _is_call_to(value.func, "replace", modules=("dataclasses", "copy"))
+ and bool(value.args)
+ and isinstance(value.args[0], ast.Attribute)
+ and _target_name(value.args[0]) == "start_trigger_args"
+ )
+
+
+def _iter_trigger_kwargs_items(call: ast.Call) -> Iterator[tuple[str | None,
ast.expr]]:
+ """
+ Yield the ``(key, value)`` pairs of a call's ``trigger_kwargs=`` argument.
+
+ Supports a ``{...}`` literal and a ``dict(...)`` call; ``**`` unpacking
yields a None key.
+ Only the ``trigger_kwargs=`` keyword is read: other keywords (``timeout``,
``next_kwargs``, ...)
+ are never rendered, and positional arguments are deliberately not matched.
+
+ :param call: The call node.
+ :return: Iterator over the key/value pairs passed as ``trigger_kwargs``.
+ """
+ for keyword in call.keywords:
+ if keyword.arg != "trigger_kwargs":
+ continue
+ if isinstance(keyword.value, ast.Dict):
+ for key, item in zip(keyword.value.keys, keyword.value.values):
+ yield (
+ (key.value if isinstance(key, ast.Constant) and
isinstance(key.value, str) else None),
+ item,
+ )
+ elif isinstance(keyword.value, ast.Call) and
_is_call_to(keyword.value.func, "dict"):
+ for inner in keyword.value.keywords:
+ yield inner.arg, inner.value
+
+
def _check_constructor_field_logic(
class_node: ast.ClassDef, template_fields: list[str], source_lines:
list[str]
) -> int:
diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt
b/scripts/ci/prek/validate_operators_init_exemptions.txt
index 117ee0886e5..45b4d646fc3 100644
--- a/scripts/ci/prek/validate_operators_init_exemptions.txt
+++ b/scripts/ci/prek/validate_operators_init_exemptions.txt
@@ -10,7 +10,6 @@
providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py::NeptuneS
providers/google/src/airflow/providers/google/cloud/operators/cloud_build.py::CloudBuildCreateBuildOperator
providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py::CloudDataTransferServiceCreateJobOperator
providers/google/src/airflow/providers/google/cloud/operators/dataproc.py::DataprocCreateClusterOperator
-providers/google/src/airflow/providers/google/cloud/operators/dataproc.py::DataprocSubmitJobOperator
providers/google/src/airflow/providers/google/cloud/operators/functions.py::CloudFunctionDeployFunctionOperator
providers/google/src/airflow/providers/google/cloud/sensors/bigquery_dts.py::BigQueryDataTransferServiceTransferRunSensor
providers/google/src/airflow/providers/google/cloud/sensors/cloud_composer.py::CloudComposerExternalTaskSensor
diff --git a/scripts/tests/ci/prek/test_validate_operators_init.py
b/scripts/tests/ci/prek/test_validate_operators_init.py
index 45302516bb1..53dc2a2db3d 100644
--- a/scripts/tests/ci/prek/test_validate_operators_init.py
+++ b/scripts/tests/ci/prek/test_validate_operators_init.py
@@ -127,6 +127,129 @@ class TestConstructorFieldLogic:
0,
id="provision-check-passed-to-a-helper",
),
+ pytest.param(
+ "self.foo = foo\nself.start_trigger_args =
StartTriggerArgs(trigger_kwargs={'foo': self.foo})",
+ 0,
+ id="verbatim-copy-into-start-trigger-args",
+ ),
+ pytest.param(
+ "self.foo = foo\n"
+ "self.start_trigger_args =
dataclasses.replace(self.start_trigger_args,
trigger_kwargs=dict(foo=self.foo))",
+ 0,
+ id="verbatim-copy-via-replace-of-start-trigger-args",
+ ),
+ pytest.param(
+ "self.foo = foo\nself.start_trigger_args =
StartTriggerArgs(trigger_kwargs={'foo': self.foo.upper()})",
+ 1,
+ id="transformation-inside-trigger-kwargs",
+ ),
+ pytest.param(
+ "self.foo = foo\nself.start_trigger_args =
StartTriggerArgs(trigger_kwargs={'bar': self.foo})",
+ 1,
+ id="trigger-kwargs-key-differs-from-field-name",
+ ),
+ pytest.param(
+ "self.foo = foo\nself.start_trigger_args =
StartTriggerArgs(trigger_kwargs={**self.foo})",
+ 1,
+ id="trigger-kwargs-unpacking",
+ ),
+ pytest.param(
+ "self.foo = foo\nself.start_trigger_args =
StartTriggerArgs(trigger_kwargs={}, timeout=foo)",
+ 1,
+
id="bare-field-name-in-a-non-rendered-start-trigger-args-field",
+ ),
+ pytest.param(
+ "self.foo = foo\n"
+ "self.start_trigger_args = StartTriggerArgs(trigger_kwargs={},
next_kwargs={'foo': self.foo})",
+ 1,
+ id="other-start-trigger-args-fields-are-not-rendered",
+ ),
+ pytest.param(
+ "self.foo = foo\nunused =
StartTriggerArgs(trigger_kwargs={'foo': self.foo})",
+ 1,
+ id="start-trigger-args-not-assigned-to-self",
+ ),
+ pytest.param(
+ "self.foo = foo\nself.other = dataclasses.replace(self.other,
foo=self.foo)",
+ 1,
+ id="replace-of-another-object-is-not-sanctioned",
+ ),
+ pytest.param(
+ "self.foo = foo\n"
+ "self.start_trigger_args = replace(self.start_trigger_args,
trigger_kwargs={'foo': self.foo})",
+ 0,
+ id="verbatim-copy-via-bare-replace",
+ ),
+ pytest.param(
+ "self.foo = foo\n"
+ "self.start_trigger_args =
copy.replace(self.start_trigger_args, trigger_kwargs={'foo': self.foo})",
+ 0,
+ id="verbatim-copy-via-copy-replace",
+ ),
+ pytest.param(
+ "self.foo = foo\nself.start_trigger_args =
factory.StartTriggerArgs(trigger_kwargs={'foo': self.foo})",
+ 1,
+ id="start-trigger-args-constructor-must-be-a-bare-name",
+ ),
+ pytest.param(
+ "self.foo = foo\n"
+ "self.start_trigger_args =
helper.replace(self.start_trigger_args, trigger_kwargs={'foo': self.foo})",
+ 1,
+ id="replace-must-come-from-dataclasses-or-copy",
+ ),
+ pytest.param(
+ "self.foo = foo\nself.start_trigger_args =
StartTriggerArgs(trigger_kwargs=helper.dict(foo=self.foo))",
+ 1,
+ id="trigger-kwargs-dict-must-be-the-builtin",
+ ),
+ pytest.param(
+ "self.foo = foo\nself.start_trigger_args =
StartTriggerArgs('cls', {'foo': self.foo})",
+ 1,
+ id="positional-trigger-kwargs-are-not-inspected",
+ ),
+ pytest.param(
+ "self.foo = foo\nself.start_trigger_args =
make_args(trigger_kwargs={'foo': self.foo})",
+ 1,
+ id="bare-start-trigger-args-constructor-name-must-match",
+ ),
+ pytest.param(
+ "self.foo = foo\nstart_trigger_args =
StartTriggerArgs(trigger_kwargs={'foo': self.foo})",
+ 1,
+ id="start-trigger-args-must-be-assigned-to-an-attribute",
+ ),
+ pytest.param(
+ "self.foo = foo\nself._sta =
StartTriggerArgs(trigger_kwargs={'foo': self.foo})",
+ 1,
+ id="start-trigger-args-attribute-name-must-match",
+ ),
+ pytest.param(
+ "self.foo = foo\n"
+ "self.start_trigger_args = dataclasses.replace(self.other,
trigger_kwargs={'foo': self.foo})",
+ 1,
+ id="replace-must-copy-start-trigger-args-itself",
+ ),
+ pytest.param(
+ "self.foo = foo\n"
+ "self.start_trigger_args =
StartTriggerArgs(trigger_kwargs=OrderedDict(foo=self.foo))",
+ 1,
+ id="bare-trigger-kwargs-dict-name-must-match",
+ ),
+ pytest.param(
+ "self.foo = foo\nself.start_trigger_args =
replace(trigger_kwargs={'foo': self.foo})",
+ 1,
+ id="replace-without-a-positional-argument",
+ ),
+ pytest.param(
+ "self.foo = foo\n"
+ "self.start_trigger_args =
a.b.replace(self.start_trigger_args, trigger_kwargs={'foo': self.foo})",
+ 1,
+ id="replace-with-a-nested-attribute-qualifier",
+ ),
+ pytest.param(
+ "self.foo = foo\nself.start_trigger_args =
BASE_ARGS[self.foo]",
+ 1,
+ id="non-call-start-trigger-args-assignment",
+ ),
],
)
def test_flags_logic_but_not_sanctioned_patterns(self, ctor_body: str,
expected: int):
@@ -156,6 +279,19 @@ class TestConstructorFieldLogic:
"""
assert _logic_findings(code, ["conf"]) == 0
+ def test_trigger_kwargs_copy_of_another_template_field_is_flagged(self):
+ # "bar" is rendered under the key "bar", so the value copied here is
the wrong field.
+ code = """
+ class MyOperator(BaseOperator):
+ template_fields = ("foo", "bar")
+
+ def __init__(self, foo=None, bar=None, **kwargs):
+ self.foo = foo
+ self.bar = bar
+ self.start_trigger_args =
StartTriggerArgs(trigger_kwargs={"foo": self.bar})
+ """
+ assert _logic_findings(code, ["foo", "bar"]) == 1
+
def test_unbound_module_name_matching_field_is_not_flagged(self):
code = """
class MyOperator(BaseOperator):
