This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 5f98b290fe7 [v3-3-test] Do not import the stored path when rebuilding
a Callback from serialized data (#70704) (#71042)
5f98b290fe7 is described below
commit 5f98b290fe7d0d2bf5dc89a2d6249d4cc7ca3ddc
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Aug 4 11:15:28 2026 +0530
[v3-3-test] Do not import the stored path when rebuilding a Callback from
serialized data (#70704) (#71042)
* Do not import the stored path when rebuilding a Callback from serialized
data
Callback.get_callback_path imports the module named by a dotted-path string
in
order to check that it resolves to a callable. That check is best effort:
the
ImportError is logged and swallowed, and the path is returned either way,
because the callable may only exist on the host that will run it.
Rebuilding a Callback from its serialized form went through the same path,
so
deserializing one imported the module named in the stored data. The path was
already checked when the Callback was first created, and reconstruction
happens
in components that never call the callback themselves, so importing there is
neither needed nor wanted.
Carry a path read back from serialized data in a private str subclass, and
return it unchanged instead of resolving it a second time. Construction
from a
Dag author's callable or dotted path is unaffected, including the dot-path
shape check, which still applies to stored paths as well.
Generated-by: Claude Opus 5 (1M context) following the guidelines at
https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions
* Keep the reason for skipping resolution at one site
The class docstring, the branch and deserialize all narrated the same
decision.
(cherry picked from commit 1baee0bf987eb220f15398e43572e1ed819a0130)
Co-authored-by: Jarek Potiuk <[email protected]>
---
task-sdk/src/airflow/sdk/definitions/callback.py | 22 ++++
.../tests/task_sdk/definitions/test_callback.py | 127 +++++++++++++++++++++
2 files changed, 149 insertions(+)
diff --git a/task-sdk/src/airflow/sdk/definitions/callback.py
b/task-sdk/src/airflow/sdk/definitions/callback.py
index 4280a43513d..355294303ab 100644
--- a/task-sdk/src/airflow/sdk/definitions/callback.py
+++ b/task-sdk/src/airflow/sdk/definitions/callback.py
@@ -28,6 +28,18 @@ from airflow.sdk._shared.module_loading import
import_string, is_valid_dotpath
log = structlog.getLogger(__name__)
+class _SerializedCallbackPath(str):
+ """
+ A callback path which was read back from serialized data.
+
+ See :meth:`Callback.get_callback_path` for what the marker buys.
+
+ :meta private:
+ """
+
+ __slots__ = ()
+
+
class Callback(ABC):
"""
Base class for Deadline Alert callbacks.
@@ -66,6 +78,14 @@ class Callback(ABC):
stripped_callback = _callback.strip()
+ if isinstance(_callback, _SerializedCallbackPath):
+ # This path was already checked when the Callback it belongs to
was created, so
+ # rebuilding that Callback keeps it as it is. Reconstruction
happens in components
+ # which never call the callback themselves, and resolving the path
there is neither
+ # needed nor wanted: the module it names is not necessarily
importable in that
+ # process, and importing it has no bearing on the path we return
either way.
+ return stripped_callback
+
try:
# The provided callback is a string which appears to be a valid
dotpath, attempt to import it.
callback = import_string(stripped_callback)
@@ -95,6 +115,8 @@ class Callback(ABC):
@classmethod
def deserialize(cls, data: dict, version):
path = data.pop("path")
+ if isinstance(path, str):
+ path = _SerializedCallbackPath(path)
return cls(callback_callable=path, **data)
@classmethod
diff --git a/task-sdk/tests/task_sdk/definitions/test_callback.py
b/task-sdk/tests/task_sdk/definitions/test_callback.py
index 8b2bdb1bdc2..11bc6365850 100644
--- a/task-sdk/tests/task_sdk/definitions/test_callback.py
+++ b/task-sdk/tests/task_sdk/definitions/test_callback.py
@@ -16,11 +16,15 @@
# under the License.
from __future__ import annotations
+import importlib
+import sys
+from pathlib import Path
from typing import cast
import pytest
from airflow.sdk._shared.module_loading import qualname
+from airflow.sdk._shared.serialization import DATA
from airflow.sdk.definitions.callback import AsyncCallback, Callback,
SyncCallback
from airflow.serialization.serde import deserialize, serialize
@@ -39,6 +43,45 @@ TEST_CALLBACK_PATH =
qualname(empty_async_callback_for_deadline_tests)
TEST_CALLBACK_KWARGS = {"arg1": "value1"}
UNIMPORTABLE_DOT_PATH = "valid.but.nonexistent.path"
+# A module which leaves a trace when its body is executed, so that a test can
tell whether
+# it was imported. Written to a temporary directory by the `callback_module`
fixture below.
+CALLBACK_MODULE_SOURCE = '''
+from pathlib import Path
+
+Path(__file__).with_suffix(".imported").touch()
+
+
+def sync_callback():
+ """A callback which can be reached by dot path once this module is
importable."""
+
+
+async def async_callback():
+ """An awaitable callback which can be reached by dot path once this module
is importable."""
+'''
+
+
[email protected]
+def callback_module(tmp_path, monkeypatch):
+ """
+ Provide a factory which makes a named module importable for the duration
of a test.
+
+ The factory returns the module name and the path of a marker file which
the module
+ creates when it is imported; the marker only exists if the module body has
been run.
+ """
+ monkeypatch.syspath_prepend(str(tmp_path))
+ created: list[str] = []
+
+ def _create(module_name: str) -> tuple[str, Path]:
+ (tmp_path / f"{module_name}.py").write_text(CALLBACK_MODULE_SOURCE)
+ importlib.invalidate_caches()
+ created.append(module_name)
+ return module_name, tmp_path / f"{module_name}.imported"
+
+ yield _create
+
+ for module_name in created:
+ sys.modules.pop(module_name, None)
+
class TestCallback:
@pytest.mark.parametrize(
@@ -229,5 +272,89 @@ class TestSyncCallback:
assert callback == deserialized
+class TestCallbackPathHandling:
+ """Cover how a callback path is treated when it is supplied, versus read
back from storage."""
+
+ def test_init_imports_the_module_named_by_the_path(self, callback_module):
+ """A Callback created from a dot path resolves it, so its author gets
feedback on it."""
+ module_name, marker =
callback_module("callback_module_resolved_on_creation")
+ path = f"{module_name}.sync_callback"
+
+ callback = SyncCallback(path)
+
+ assert marker.exists(), "the module named by the path should have been
imported"
+ assert callback.path == path
+
+ def
test_deserialize_does_not_import_the_module_named_by_the_stored_path(self,
callback_module):
+ """Rebuilding a Callback keeps the stored path without resolving it
again."""
+ module_name, marker =
callback_module("callback_module_not_resolved_on_deserialize")
+ path = f"{module_name}.sync_callback"
+
+ callback = SyncCallback.deserialize({"path": path, "kwargs": {},
"executor": None}, 0)
+
+ assert not marker.exists(), "the module named by the stored path
should not have been imported"
+ assert module_name not in sys.modules
+ assert callback.path == path
+ assert type(callback.path) is str
+
+ def
test_deserialize_async_does_not_import_the_module_named_by_the_stored_path(self,
callback_module):
+ module_name, marker =
callback_module("async_callback_module_not_resolved_on_deserialize")
+ path = f"{module_name}.async_callback"
+
+ callback = AsyncCallback.deserialize({"path": path, "kwargs": {}}, 0)
+
+ assert not marker.exists(), "the module named by the stored path
should not have been imported"
+ assert module_name not in sys.modules
+ assert callback.path == path
+
+ def
test_serde_deserialize_does_not_import_the_module_named_by_the_stored_path(self,
callback_module):
+ """The same holds when the Callback is rebuilt through serde, as it is
from stored data."""
+ module_name, marker =
callback_module("callback_module_not_resolved_through_serde")
+ path = f"{module_name}.sync_callback"
+
+ serialized = serialize(SyncCallback(TEST_CALLBACK_PATH,
kwargs=TEST_CALLBACK_KWARGS))
+ serialized[DATA]["path"] = path
+
+ deserialized = cast("Callback", deserialize(serialized))
+
+ assert not marker.exists(), "the module named by the stored path
should not have been imported"
+ assert module_name not in sys.modules
+ assert deserialized.path == path
+
+ @pytest.mark.parametrize(
+ "path",
+ [
+ pytest.param("not a dot path", id="not_a_dot_path"),
+ pytest.param("", id="empty_string"),
+ pytest.param(None, id="none"),
+ pytest.param(42, id="not_a_string"),
+ ],
+ )
+ def test_deserialize_rejects_a_path_which_is_not_a_dot_path(self, path):
+ """A stored value which is not shaped like a dot path is still
rejected."""
+ with pytest.raises(ImportError, match="doesn't look like a valid dot
path."):
+ SyncCallback.deserialize({"path": path, "kwargs": {}, "executor":
None}, 0)
+
+ def
test_deserialize_keeps_a_stored_path_which_no_longer_points_at_a_callable(self):
+ """
+ The stored path is taken as it was stored.
+
+ Unlike a path handed to the constructor, it is not checked against
what it currently
+ resolves to; that check belongs to the moment the Callback is created.
+ """
+ callback = SyncCallback.deserialize({"path": "os.path", "kwargs": {},
"executor": None}, 0)
+
+ assert callback.path == "os.path"
+
+ def test_deserialize_round_trip_keeps_kwargs_and_executor(self):
+ callback = SyncCallback(TEST_CALLBACK_PATH,
kwargs=TEST_CALLBACK_KWARGS, executor="local")
+
+ deserialized = cast("SyncCallback", deserialize(serialize(callback)))
+
+ assert deserialized == callback
+ assert deserialized.kwargs == TEST_CALLBACK_KWARGS
+ assert deserialized.executor == "local"
+
+
# While DeadlineReference lives in the SDK package, the unit tests to confirm
it
# works need database access so they live in the models/test_deadline.py
module.
