vatsrahul1001 commented on code in PR #74842:
URL: https://github.com/apache/airflow/pull/74842#discussion_r4306143502
##########
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.
Review Comment:
Optional: per our newsfragment guidelines these are for major or breaking
changes. This is hardening that still accepts every documented use, so it can
probably be dropped.
##########
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:
Nit: the `os.getcwd` patch only matters for the `sync-callable` case, where
`import_string("os.getcwd")` returns the mock because the patch is active
before the first `anext`. That is subtle. Patching only for that case, or using
a local sync function with a spy, would make the intent clearer.
##########
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:
Optional: this duplicates `AsyncCallback.verify_callable` and extends it
with the `async def __call__` case, which the SDK check rejects. A path to an
async callable object would be refused when the callback is defined but
accepted here. Either add that case to the SDK check too, or drop it here, so
there is a single source of truth.
--
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]
