This is an automated email from the ASF dual-hosted git repository.
potiuk 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 57fb394afe8 Validate CallbackTrigger callback path before running the
callback (#74842)
57fb394afe8 is described below
commit 57fb394afe88a077074914d76db572f68856c0bb
Author: Jarek Potiuk <[email protected]>
AuthorDate: Mon Oct 19 16:30:33 2026 +0200
Validate CallbackTrigger callback path before running the callback (#74842)
* Validate CallbackTrigger callback path before running the callback
The callback path resolved by CallbackTrigger must now be a coroutine
function or an awaitable callable (such as an async notifier class),
matching what AsyncCallback accepts. Anything else fails the callback
before it is called, the same way trigger classpaths are validated.
Generated-by: Claude Opus 5 following the guidelines at
https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions
* Rename newsfragment to the PR number
Generated-by: Claude Opus 5 following the guidelines at
https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions
* Accept the same triggerer callbacks the Task SDK accepts
The triggerer also accepted callable objects with an async __call__, which
AsyncCallback rejects when the callback is defined, so the two checks
disagreed. Core cannot import the SDK check, so the triggerer now applies the
same rule locally.
Generated-by: Claude Opus 5
---
airflow-core/newsfragments/74842.improvement.rst | 1 +
airflow-core/src/airflow/triggers/callback.py | 27 +++++
airflow-core/tests/unit/triggers/test_callback.py | 114 +++++++++++++++++++++-
3 files changed, 141 insertions(+), 1 deletion(-)
diff --git a/airflow-core/newsfragments/74842.improvement.rst
b/airflow-core/newsfragments/74842.improvement.rst
new file mode 100644
index 00000000000..48302129c64
--- /dev/null
+++ b/airflow-core/newsfragments/74842.improvement.rst
@@ -0,0 +1 @@
+The triggerer now validates that a triggerer callback path resolves to a
coroutine function or an awaitable callable (for example an async notifier
class) before running it, the same way trigger classpaths are validated. Other
targets fail the callback without being called.
diff --git a/airflow-core/src/airflow/triggers/callback.py
b/airflow-core/src/airflow/triggers/callback.py
index 7645c42e980..45739f48226 100644
--- a/airflow-core/src/airflow/triggers/callback.py
+++ b/airflow-core/src/airflow/triggers/callback.py
@@ -17,6 +17,7 @@
from __future__ import annotations
+import inspect
import logging
import traceback
from collections.abc import AsyncIterator
@@ -32,6 +33,29 @@ PAYLOAD_STATUS_KEY = "state"
PAYLOAD_BODY_KEY = "body"
+class InvalidCallbackPathError(TypeError):
+ """Raised when a callback path does not resolve to a usable triggerer
callback."""
+
+
+def _validate_callback(callback: Any, callback_path: str) -> None:
+ """
+ Check that the object resolved from ``callback_path`` can be used as a
triggerer callback.
+
+ It accepts exactly what ``AsyncCallback.verify_callable`` in the Task SDK
accepts when the
+ callback is defined, and is run again because the path is read back from
the database. Core
+ cannot import that check from the SDK, so keep the two in step. The path
is validated before
+ the callable is invoked, the same way trigger classpaths are validated
before a trigger is
+ instantiated.
+ """
+ if not callable(callback):
+ raise InvalidCallbackPathError(f"The callback path {callback_path!r}
does not resolve to a callable.")
+ if not (inspect.iscoroutinefunction(callback) or hasattr(callback,
"__await__")):
+ raise InvalidCallbackPathError(
+ f"The callback path {callback_path!r} does not resolve to a
coroutine function or "
+ f"an awaitable callable; refusing to run it on the triggerer."
+ )
+
+
class CallbackTrigger(BaseTrigger):
"""Trigger that executes a callback function asynchronously."""
@@ -62,6 +86,7 @@ class CallbackTrigger(BaseTrigger):
try:
yield TriggerEvent({PAYLOAD_STATUS_KEY: CallbackState.RUNNING})
callback = import_string(self.callback_path)
+ _validate_callback(callback, self.callback_path)
# TODO: get full context and run template rendering. Right now, a
simple context is included in `callback_kwargs`
context = self.callback_kwargs.pop("context", None)
@@ -75,6 +100,8 @@ class CallbackTrigger(BaseTrigger):
except Exception as e:
if isinstance(e, ImportError):
message = "Failed to import the callable on the triggerer"
+ elif isinstance(e, InvalidCallbackPathError):
+ message = "The callback path does not resolve to a valid
callback"
elif isinstance(e, TypeError) and "await" in str(e):
message = "Failed to run the callable because it's not
awaitable"
else:
diff --git a/airflow-core/tests/unit/triggers/test_callback.py
b/airflow-core/tests/unit/triggers/test_callback.py
index 6707b9ada2a..18a7fec8c65 100644
--- a/airflow-core/tests/unit/triggers/test_callback.py
+++ b/airflow-core/tests/unit/triggers/test_callback.py
@@ -17,13 +17,20 @@
from __future__ import annotations
+import os
from unittest import mock
import pytest
from airflow.models.callback import CallbackState
from airflow.sdk import BaseNotifier
-from airflow.triggers.callback import PAYLOAD_BODY_KEY, PAYLOAD_STATUS_KEY,
CallbackTrigger
+from airflow.triggers.callback import (
+ PAYLOAD_BODY_KEY,
+ PAYLOAD_STATUS_KEY,
+ CallbackTrigger,
+ InvalidCallbackPathError,
+ _validate_callback,
+)
TEST_MESSAGE = "test_message"
TEST_CALLBACK_PATH = "classpath.test_callback"
@@ -44,6 +51,57 @@ class ExampleAsyncNotifier(BaseNotifier):
return f"Sync notification: {self.message}, context: {context}"
+async def example_async_callback(message, context=None):
+ return f"Async callback: {message}"
+
+
+def example_sync_callback(message, context=None):
+ return f"Sync callback: {message}"
+
+
+class ExampleAsyncCallableObject:
+ async def __call__(self, message, context=None):
+ return f"Async callable object: {message}"
+
+
+class ExamplePlainClass:
+ def __init__(self, message, context=None):
+ self.message = message
+
+
+class TestValidateCallback:
+ @pytest.mark.parametrize(
+ "callback",
+ [
+ pytest.param(example_async_callback, id="coroutine-function"),
+ pytest.param(ExampleAsyncNotifier, id="async-notifier-class"),
+ pytest.param(mock.AsyncMock(), id="async-mock"),
+ ],
+ )
+ def test_valid_callbacks_pass(self, callback):
+ _validate_callback(callback, "some.callback.path")
+
+ @pytest.mark.parametrize(
+ ("callback", "match"),
+ [
+ pytest.param(os, "does not resolve to a callable", id="module"),
+ pytest.param("not-a-callable", "does not resolve to a callable",
id="string"),
+ pytest.param(42, "does not resolve to a callable", id="int"),
+ pytest.param(example_sync_callback, "coroutine function or an
awaitable", id="sync-function"),
+ pytest.param(os.getcwd, "coroutine function or an awaitable",
id="sync-builtin"),
+ pytest.param(ExamplePlainClass, "coroutine function or an
awaitable", id="plain-class"),
+ pytest.param(
+ ExampleAsyncCallableObject(),
+ "coroutine function or an awaitable",
+ id="async-callable-object-rejected-like-the-sdk",
+ ),
+ ],
+ )
+ def test_invalid_callbacks_rejected(self, callback, match):
+ with pytest.raises(InvalidCallbackPathError, match=match):
+ _validate_callback(callback, "some.callback.path")
+
+
class TestCallbackTrigger:
@pytest.fixture
def trigger(self):
@@ -151,3 +209,57 @@ class TestCallbackTrigger:
mock_callback.assert_called_once_with(**TEST_CALLBACK_KWARGS)
assert failure_event.payload[PAYLOAD_STATUS_KEY] ==
CallbackState.FAILED
assert all(s in failure_event.payload[PAYLOAD_BODY_KEY] for s in
["raise", "RuntimeError", exc_msg])
+
+ @pytest.mark.asyncio
+ async def test_run_with_real_coroutine_function_path(self):
+ trigger = CallbackTrigger(
+ callback_path=f"{__name__}.example_async_callback",
+ callback_kwargs={"message": TEST_MESSAGE},
+ )
+ trigger_gen = trigger.run()
+
+ assert (await anext(trigger_gen)).payload[PAYLOAD_STATUS_KEY] ==
CallbackState.RUNNING
+ success_event = await anext(trigger_gen)
+ assert success_event.payload[PAYLOAD_STATUS_KEY] ==
CallbackState.SUCCESS
+ assert success_event.payload[PAYLOAD_BODY_KEY] == f"Async callback:
{TEST_MESSAGE}"
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ "callback_path",
+ [
+ pytest.param("os.path", id="module"),
+ pytest.param("os.sep", id="non-callable-attribute"),
+ pytest.param(f"{__name__}.ExamplePlainClass", id="plain-class"),
+ ],
+ )
+ async def test_run_fails_for_invalid_callback_path(self, callback_path):
+ trigger = CallbackTrigger(callback_path=callback_path,
callback_kwargs={})
+ trigger_gen = trigger.run()
+
+ assert (await anext(trigger_gen)).payload[PAYLOAD_STATUS_KEY] ==
CallbackState.RUNNING
+ failure_event = await anext(trigger_gen)
+
+ assert failure_event.payload[PAYLOAD_STATUS_KEY] ==
CallbackState.FAILED
+ assert (
+ "The callback path does not resolve to a valid callback"
+ in failure_event.payload[PAYLOAD_BODY_KEY]
+ )
+ assert "InvalidCallbackPathError" in
failure_event.payload[PAYLOAD_BODY_KEY]
+
+ @pytest.mark.asyncio
+ async def test_run_refuses_a_sync_callable_without_calling_it(self):
+ calls = []
+
+ def sync_spy(*args, **kwargs):
+ calls.append((args, kwargs))
+
+ trigger =
CallbackTrigger(callback_path=f"{__name__}.example_sync_callback",
callback_kwargs={})
+ trigger_gen = trigger.run()
+
+ with mock.patch(f"{__name__}.example_sync_callback", sync_spy):
+ assert (await anext(trigger_gen)).payload[PAYLOAD_STATUS_KEY] ==
CallbackState.RUNNING
+ failure_event = await anext(trigger_gen)
+
+ assert calls == []
+ assert failure_event.payload[PAYLOAD_STATUS_KEY] ==
CallbackState.FAILED
+ assert "InvalidCallbackPathError" in
failure_event.payload[PAYLOAD_BODY_KEY]
