Skip to content

Commit fbb200e

Browse files
kaxilkaxil
authored
Fix unnecessary DAG version churn when DAG file paths change (#60799)
1 parent 2abd309 commit fbb200e

4 files changed

Lines changed: 133 additions & 13 deletions

File tree

‎airflow-core/src/airflow/serialization/serialized_objects.py‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1035,11 +1035,13 @@ def serialize_mapped_operator(cls, op: MappedOperator) -> dict[str, Any]:
10351035
continue
10361036
serialized_op["partial_kwargs"].update({k: cls.serialize(v)})
10371037

1038-
# we want to store python_callable_name, not python_callable
1038+
# Store python_callable_name instead of python_callable.
1039+
# exclude_module=True ensures stable names across bundle version changes.
10391040
python_callable = op.partial_kwargs.get("python_callable", None)
10401041
if python_callable:
1041-
callable_name = qualname(python_callable)
1042-
serialized_op["partial_kwargs"]["python_callable_name"] = callable_name
1042+
serialized_op["partial_kwargs"]["python_callable_name"] = qualname(
1043+
python_callable, exclude_module=True
1044+
)
10431045
del serialized_op["partial_kwargs"]["python_callable"]
10441046

10451047
serialized_op["_is_mapped"] = True
@@ -1060,11 +1062,11 @@ def _serialize_node(cls, op: SdkOperator) -> dict[str, Any]:
10601062
if attr in serialize_op:
10611063
del serialize_op[attr]
10621064

1063-
# Detect if there's a change in python callable name
1065+
# Store python_callable_name for change detection.
1066+
# exclude_module=True ensures stable names across bundle version changes.
10641067
python_callable = getattr(op, "python_callable", None)
10651068
if python_callable:
1066-
callable_name = qualname(python_callable)
1067-
serialize_op["python_callable_name"] = callable_name
1069+
serialize_op["python_callable_name"] = qualname(python_callable, exclude_module=True)
10681070

10691071
serialize_op["task_type"] = getattr(op, "task_type", type(op).__name__)
10701072
serialize_op["_task_module"] = getattr(op, "_task_module", type(op).__module__)

‎airflow-core/tests/unit/serialization/test_dag_serialization.py‎

Lines changed: 36 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import contextlib
2323
import copy
2424
import dataclasses
25+
import functools
2526
import importlib
2627
import importlib.util
2728
import json
@@ -2934,7 +2935,7 @@ def x(arg1, arg2, arg3):
29342935
},
29352936
"_disallow_kwargs_override": False,
29362937
"_expand_input_attr": "op_kwargs_expand_input",
2937-
"python_callable_name": qualname(x),
2938+
"python_callable_name": "test_taskflow_expand_serde.<locals>.x",
29382939
}
29392940

29402941
deserialized = BaseSerialization.deserialize(serialized)
@@ -3001,7 +3002,7 @@ def x(arg1, arg2, arg3):
30013002
"_task_module": "airflow.providers.standard.decorators.python",
30023003
"task_type": "_PythonDecoratedOperator",
30033004
"_operator_name": "@task",
3004-
"python_callable_name": qualname(x),
3005+
"python_callable_name": "test_taskflow_expand_kwargs_serde.<locals>.x",
30053006
"partial_kwargs": {
30063007
"op_args": [],
30073008
"op_kwargs": {
@@ -3172,11 +3173,42 @@ def test_python_callable_in_partial_kwargs():
31723173

31733174
serialized = OperatorSerialization.serialize_mapped_operator(operator)
31743175
assert "python_callable" not in serialized["partial_kwargs"]
3175-
assert serialized["partial_kwargs"]["python_callable_name"] == qualname(empty_function)
3176+
assert serialized["partial_kwargs"]["python_callable_name"] == "empty_function"
31763177

31773178
deserialized = OperatorSerialization.deserialize_operator(serialized)
31783179
assert "python_callable" not in deserialized.partial_kwargs
3179-
assert deserialized.partial_kwargs["python_callable_name"] == qualname(empty_function)
3180+
assert deserialized.partial_kwargs["python_callable_name"] == "empty_function"
3181+
3182+
3183+
def test_python_callable_name_uses_qualname_exclude_module():
3184+
"""Test python_callable_name is stable across bundle version changes."""
3185+
from airflow.providers.standard.operators.python import PythonOperator
3186+
3187+
# Module-level function
3188+
op1 = PythonOperator(task_id="task1", python_callable=empty_function)
3189+
serialized1 = OperatorSerialization.serialize_operator(op1)
3190+
assert serialized1["python_callable_name"] == "empty_function"
3191+
3192+
# Nested function
3193+
def outer():
3194+
def inner():
3195+
pass
3196+
3197+
return inner
3198+
3199+
inner_func = outer()
3200+
op2 = PythonOperator(task_id="task2", python_callable=inner_func)
3201+
serialized2 = OperatorSerialization.serialize_operator(op2)
3202+
assert (
3203+
serialized2["python_callable_name"]
3204+
== "test_python_callable_name_uses_qualname_exclude_module.<locals>.outer.<locals>.inner"
3205+
)
3206+
3207+
# functools.partial
3208+
partial_func = functools.partial(empty_function, x=1)
3209+
op3 = PythonOperator(task_id="task3", python_callable=partial_func)
3210+
serialized3 = OperatorSerialization.serialize_operator(op3)
3211+
assert serialized3["python_callable_name"] == "empty_function"
31803212

31813213

31823214
def test_handle_v1_serdag():

‎shared/module_loading/src/airflow_shared/module_loading/__init__.py‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -63,9 +63,24 @@ def import_string(dotted_path: str):
6363
raise ImportError(f'Module "{module_path}" does not define a "{class_name}" attribute/class')
6464

6565

66-
def qualname(o: object | Callable, use_qualname: bool = False) -> str:
67-
"""Convert an attribute/class/callable to a string importable by ``import_string``."""
66+
def qualname(o: object | Callable, use_qualname: bool = False, exclude_module: bool = False) -> str:
67+
"""
68+
Convert an attribute/class/callable to a string.
69+
70+
By default, returns a string importable by ``import_string`` (includes module path).
71+
With exclude_module=True, returns only the qualified name without module prefix,
72+
useful for stable identification across deployments where module paths may vary.
73+
"""
6874
if callable(o) and hasattr(o, "__module__"):
75+
if exclude_module:
76+
if hasattr(o, "__qualname__"):
77+
return o.__qualname__
78+
if hasattr(o, "__name__"):
79+
return o.__name__
80+
# Handle functools.partial objects specifically (not just any object with 'func' attr)
81+
if isinstance(o, functools.partial):
82+
return qualname(o.func, exclude_module=True)
83+
return type(o).__qualname__
6984
if use_qualname and hasattr(o, "__qualname__"):
7085
return f"{o.__module__}.{o.__qualname__}"
7186
if hasattr(o, "__name__"):
@@ -79,6 +94,9 @@ def qualname(o: object | Callable, use_qualname: bool = False) -> str:
7994
name = cls.__qualname__
8095
module = cls.__module__
8196

97+
if exclude_module:
98+
return name
99+
82100
if module and module != "__builtin__":
83101
return f"{module}.{name}"
84102

‎shared/module_loading/tests/module_loading/test_module_loading.py‎

Lines changed: 69 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,15 +17,21 @@
1717
# under the License.
1818
from __future__ import annotations
1919

20+
import functools
21+
2022
import pytest
2123

22-
from airflow_shared.module_loading import import_string, is_valid_dotpath
24+
from airflow_shared.module_loading import import_string, is_valid_dotpath, qualname
2325

2426

2527
def _import_string():
2628
pass
2729

2830

31+
def _sample_function():
32+
pass
33+
34+
2935
class TestModuleImport:
3036
def test_import_string(self):
3137
cls = import_string("module_loading.test_module_loading._import_string")
@@ -56,3 +62,65 @@ class TestModuleLoading:
5662
)
5763
def test_is_valid_dotpath(self, path, expected):
5864
assert is_valid_dotpath(path) == expected
65+
66+
67+
class TestQualname:
68+
def test_qualname_default_includes_module(self):
69+
"""Test that qualname() by default includes the module path."""
70+
result = qualname(_sample_function)
71+
assert result == "module_loading.test_module_loading._sample_function"
72+
73+
def test_qualname_exclude_module_simple_function(self):
74+
"""Test that exclude_module=True returns only the function name."""
75+
result = qualname(_sample_function, exclude_module=True)
76+
assert result == "_sample_function"
77+
78+
def test_qualname_exclude_module_nested_function(self):
79+
"""Test that exclude_module=True works with nested functions."""
80+
81+
def outer():
82+
def inner():
83+
pass
84+
85+
return inner
86+
87+
inner_func = outer()
88+
result = qualname(inner_func, exclude_module=True)
89+
assert (
90+
result
91+
== "TestQualname.test_qualname_exclude_module_nested_function.<locals>.outer.<locals>.inner"
92+
)
93+
94+
def test_qualname_exclude_module_functools_partial(self):
95+
"""Test that exclude_module=True handles functools.partial correctly."""
96+
97+
def base_func(x, y):
98+
pass
99+
100+
partial_func = functools.partial(base_func, x=1)
101+
result = qualname(partial_func, exclude_module=True)
102+
assert result == "TestQualname.test_qualname_exclude_module_functools_partial.<locals>.base_func"
103+
104+
def test_qualname_exclude_module_class(self):
105+
"""Test that exclude_module=True works with classes."""
106+
107+
class MyClass:
108+
pass
109+
110+
result = qualname(MyClass, exclude_module=True)
111+
assert result == "TestQualname.test_qualname_exclude_module_class.<locals>.MyClass"
112+
113+
def test_qualname_exclude_module_instance(self):
114+
"""Test that exclude_module=True works with class instances."""
115+
116+
class MyClass:
117+
pass
118+
119+
instance = MyClass()
120+
result = qualname(instance, exclude_module=True)
121+
assert result == "TestQualname.test_qualname_exclude_module_instance.<locals>.MyClass"
122+
123+
def test_qualname_use_qualname_still_includes_module(self):
124+
"""Test that use_qualname=True still includes module prefix."""
125+
result = qualname(_sample_function, use_qualname=True)
126+
assert result == "module_loading.test_module_loading._sample_function"

0 commit comments

Comments
 (0)