potiuk commented on code in PR #74842:
URL: https://github.com/apache/airflow/pull/74842#discussion_r4313056578
##########
airflow-core/src/airflow/triggers/callback.py:
##########
@@ -32,6 +33,33 @@
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.
+
+ Triggerer callbacks (``AsyncCallback``) are coroutine functions or
awaitable callables such as
+ notifier classes, which is also what ``AsyncCallback`` checks when the
callback is defined.
+ Callable objects with an ``async def __call__`` are accepted as well.
+ The callback path is validated here 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__")
+ or inspect.iscoroutinefunction(type(callback).__call__)
Review Comment:
Went with the second option. airflow-core can't import from the Task SDK
(`check-sdk-imports-in-core`), so the triggerer keeps its own check, but it now
accepts exactly what `AsyncCallback.verify_callable` accepts. The `async def
__call__` case is gone, and the docstring points at the SDK check so the two
stay in step. Fixed in e668178b6d.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
airflow-core/tests/unit/triggers/test_callback.py:
##########
@@ -151,3 +205,42 @@ async def test_run_failure(self, trigger,
mock_import_string):
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("os.getcwd", id="sync-callable"),
+ 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()
+
+ with mock.patch("os.getcwd") as mock_getcwd:
Review Comment:
Done in e668178b6d. The `os.getcwd` patch is gone; the sync case is now its
own test that patches a module-level sync function with a recording spy and
asserts it was never called.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
