WIP: Serialized object property

airflow/serialization/serialized_objects.py

@@ -54,6 +54,7 @@
 from airflow.timetables.base import Timetable
 from airflow.utils.code_utils import get_python_source
 from airflow.utils.docs import get_docs_url
+from airflow.utils.helpers import resolve_property_value
 from airflow.utils.module_loading import as_importable_string, import_string
 from airflow.utils.operator_resources import Resources
 from airflow.utils.task_group import TaskGroup
@@ -349,6 +350,8 @@ def _serialize(cls, var: Any) -> Any:  # Unfortunately there is no support for r
         (3) Operator has a special field CLASS to record the original class
             name for displaying in UI.
         """
+        var = resolve_property_value(cls, var)
+
         if cls._is_primitive(var):
             # enum.IntEnum is an int instance, it causes json dumps error so we use its value.
             if isinstance(var, enum.Enum):
@@ -690,7 +693,7 @@ def _serialize_node(cls, op: Union[BaseOperator, MappedOperator], include_deps:
 
         if op.operator_extra_links:
             serialize_op['_operator_extra_links'] = cls._serialize_operator_extra_links(
-                op.operator_extra_links
+                resolve_property_value(op, op.operator_extra_links)
             )
 
         if include_deps:

airflow/utils/helpers.py

@@ -376,3 +376,16 @@ def is_empty(x):
         return new_list
     else:
         return val
+
+
+def resolve_property_value(obj, prop):
+    """Return class property value.
+
+    :param obj: Reference to class.
+    :param prop: Reference to class property.
+    :returns: If ``prop`` property than return property value,
+        otherwise it returns class attribute value.
+    """
+    if isinstance(prop, property):
+        return prop.fget(obj)
+    return prop

tests/serialization/test_dag_serialization.py

@@ -52,7 +52,7 @@
 from airflow.utils.context import Context
 from airflow.utils.operator_resources import Resources
 from airflow.utils.task_group import TaskGroup
-from tests.test_utils.mock_operators import CustomOperator, GoogleLink, MockOperator
+from tests.test_utils.mock_operators import CustomOperator, Dummy4TestOperator, GoogleLink, MockOperator
 from tests.test_utils.timetables import CustomSerializationTimetable, cron_timetable, delta_timetable
 
 executor_config_pod = k8s.V1Pod(
@@ -2060,3 +2060,39 @@ class MyDummyOperator(DummyOperator):
         'template_fields': [],
         'template_fields_renderers': {},
     }
+
+
+def test_operator_partial_property_links_serde():
+    with DAG("test-dag", start_date=datetime(2020, 1, 1)):
+        mapped = Dummy4TestOperator.partial(task_id='mapped_task').expand(arg1=[1, 2, 3])
+
+    serialized = SerializedBaseOperator._serialize(mapped)
+
+    assert serialized == {
+        '_is_empty': False,
+        '_is_mapped': True,
+        '_task_module': 'tests.test_utils.mock_operators',
+        '_task_type': 'Dummy4TestOperator',
+        '_operator_extra_links': [{'tests.test_utils.mock_operators.AirflowLink': {}}],
+        'downstream_task_ids': [],
+        'expand_input': {
+            "type": "dict-of-lists",
+            "value": {
+                "__type": "dict",
+                "__var": {'arg1': [1, 2, 3]},
+            },
+        },
+        'partial_kwargs': {},
+        'task_id': 'mapped_task',
+        'template_fields': [],
+        'template_ext': [],
+        'template_fields_renderers': {},
+        'operator_extra_links': ['AirflowLink()'],
+        'ui_color': '#fff',
+        'ui_fgcolor': '#000',
+        "_disallow_kwargs_override": False,
+        '_expand_input_attr': 'expand_input',
+    }
+
+    op = SerializedBaseOperator.deserialize_operator(serialized)
+    assert op.deps is MappedOperator.deps_for(BaseOperator)

tests/test_utils/mock_operators.py

@@ -67,6 +67,20 @@ class Dummy3TestOperator(BaseOperator):
     operator_extra_links = ()
 
 
+class Dummy4TestOperator(BaseOperator):
+    """
+    Example of an Operator that has an extra operator link as property
+    """
+
+    @property
+    def operator_extra_links(self):
+        return (AirflowLink(),)
+
+    def __init__(self, arg1=42, **kwargs):
+        super().__init__(**kwargs)
+        self.arg1 = arg1
+
+
 @attr.s(auto_attribs=True)
 class CustomBaseIndexOpLink(BaseOperatorLink):
     index: int = attr.ib()

tests/utils/test_helpers.py

@@ -28,6 +28,7 @@
     exactly_one,
     merge_dicts,
     prune_dict,
+    resolve_property_value,
     validate_group_key,
     validate_key,
 )
@@ -321,3 +322,28 @@ def test_prune_dict(self, mode, expected):
         d1 = {'a': None, 'b': '', 'c': 'hi', 'd': l1}
         d2 = {'a': None, 'b': '', 'c': d1, 'd': l1, 'e': [None, '', 0, d1, l1, ['']]}
         assert prune_dict(d2, mode=mode) == expected
+
+    def test_resolve_property_value_property(self):
+        class MockClass:
+            @property
+            def mock_property(self):
+                return "mock-property-value"
+
+        assert isinstance(MockClass.mock_property, property)
+        assert MockClass.mock_property != "mock-property-value"
+        assert resolve_property_value(MockClass, MockClass.mock_property) == "mock-property-value"
+        mock_class_instance = MockClass()
+        assert (
+            resolve_property_value(mock_class_instance, mock_class_instance.mock_property)
+            == "mock-property-value"
+        )
+
+    def test_resolve_property_value_attr(self):
+        class MockClass:
+            mock_attr = "mock-attr-value"
+
+        assert not isinstance(MockClass.mock_attr, property)
+        assert MockClass.mock_attr == "mock-attr-value"
+        assert resolve_property_value(MockClass, MockClass.mock_attr) == "mock-attr-value"
+        mock_class_instance = MockClass()
+        assert resolve_property_value(mock_class_instance, mock_class_instance.mock_attr) == "mock-attr-value"