fix task group serialize logics

Author: wjddn279

airflow-core/src/airflow/serialization/definitions/baseoperator.py

@@ -326,8 +326,8 @@ def __getattr__(self, name):
         # For regular attributes, use task_type in the error message
         raise AttributeError(f"'{self.task_type}' object has no attribute '{name}'")
 
-    def serialize_for_task_group(self) -> tuple[DagAttributeTypes, Any]:
-        return DagAttributeTypes.OP, self.task_id
+    def serialize_for_task_group(self) -> list:
+        return [DagAttributeTypes.OP.value, self.task_id]
 
     def expand_start_from_trigger(self, *, context: Context) -> bool:
         """

airflow-core/src/airflow/serialization/definitions/mappedoperator.py

@@ -413,9 +413,9 @@ def get_extra_links(self, ti: TaskInstance, name: str) -> str | None:
             return None
         return link.get_link(self, ti_key=ti.key)
 
-    def serialize_for_task_group(self) -> tuple[DagAttributeTypes, Any]:
+    def serialize_for_task_group(self) -> list:
         """Implement DAGNode."""
-        return DagAttributeTypes.OP, self.task_id
+        return [DagAttributeTypes.OP.value, self.task_id]
 
     # TODO (GH-52141): Copied from sdk. Find a better place for this to live in.
     def _get_specified_expand_input(self) -> SchedulerExpandInput:

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

@@ -2092,7 +2092,7 @@ def serialize_task_group(cls, task_group: TaskGroup) -> dict[str, Any] | None:
             "ui_color": task_group.ui_color,
             "ui_fgcolor": task_group.ui_fgcolor,
             "children": {
-                label: child.match_serialize_task_group_form() for label, child in task_group.children.items()
+                label: child.serialize_for_task_group() for label, child in task_group.children.items()
             },
             "upstream_group_ids": cls.serialize(sorted(task_group.upstream_group_ids)),
             "downstream_group_ids": cls.serialize(sorted(task_group.downstream_group_ids)),

task-sdk/src/airflow/sdk/bases/operator.py

@@ -95,7 +95,6 @@ def db_safe_priority(priority_weight: int) -> int:
     from airflow.sdk.definitions.operator_resources import Resources
     from airflow.sdk.definitions.taskgroup import TaskGroup
     from airflow.sdk.definitions.xcom_arg import XComArg
-    from airflow.serialization.enums import DagAttributeTypes
     from airflow.task.priority_strategy import PriorityWeightStrategy
     from airflow.triggers.base import BaseTrigger, StartTriggerArgs
 
@@ -1552,11 +1551,11 @@ def prepare_for_execution(self) -> Self:
         other._lock_for_execution = True
         return other
 
-    def serialize_for_task_group(self) -> tuple[DagAttributeTypes, Any]:
+    def serialize_for_task_group(self) -> list:
         """Serialize; required by DAGNode."""
         from airflow.serialization.enums import DagAttributeTypes
 
-        return DagAttributeTypes.OP, self.task_id
+        return [DagAttributeTypes.OP.value, self.task_id]
 
     def unmap(self, resolve: None | Mapping[str, Any]) -> Self:
         """

task-sdk/src/airflow/sdk/definitions/_internal/node.py

@@ -21,7 +21,7 @@
 from abc import ABCMeta, abstractmethod
 from collections.abc import Sequence
 from datetime import datetime
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 from airflow.sdk._shared.dagnode.node import GenericDAGNode
 from airflow.sdk.definitions._internal.mixins import DependencyMixin
@@ -31,7 +31,6 @@
     from airflow.sdk.definitions.edges import EdgeModifier
     from airflow.sdk.definitions.taskgroup import TaskGroup  # noqa: F401
     from airflow.sdk.types import Operator  # noqa: F401
-    from airflow.serialization.enums import DagAttributeTypes
 
 
 KEY_REGEX = re.compile(r"^[\w.-]+$")
@@ -157,18 +156,6 @@ def set_upstream(
         """Set a node (or nodes) to be directly upstream from the current node."""
         self._set_relatives(task_or_task_list, upstream=True, edge_modifier=edge_modifier)
 
-    def serialize_for_task_group(self) -> tuple[DagAttributeTypes, Any]:
+    def serialize_for_task_group(self) -> list:
         """Serialize a task group's content; used by TaskGroupSerialization."""
         raise NotImplementedError()
-
-    def match_serialize_task_group_form(self) -> list:
-        """
-        Match the serialized task_group format modified during inter-process communication.
-
-        The serialized task_group from dag-process gets modified during inter-process communication.
-        (<DagAttributeTypes.OP: 'operator'>, 'task_id') -> ['operator', 'task_id']
-
-        This function aligns the values to match the modified state after IPC.
-        """
-        serialized_task_group = self.serialize_for_task_group()
-        return [serialized_task_group[0].value, serialized_task_group[1]]

task-sdk/src/airflow/sdk/definitions/mappedoperator.py

@@ -713,9 +713,9 @@ def output(self) -> XComArg:
 
         return XComArg(operator=self)
 
-    def serialize_for_task_group(self) -> tuple[DagAttributeTypes, Any]:
+    def serialize_for_task_group(self) -> list:
         """Implement DAGNode."""
-        return DagAttributeTypes.OP, self.task_id
+        return [DagAttributeTypes.OP.value, self.task_id]
 
     def _expand_mapped_kwargs(self, context: Mapping[str, Any]) -> tuple[Mapping[str, Any], set[int]]:
         """

task-sdk/src/airflow/sdk/definitions/taskgroup.py

@@ -43,7 +43,6 @@
     from airflow.sdk.definitions.dag import DAG
     from airflow.sdk.definitions.edges import EdgeModifier
     from airflow.sdk.types import Operator
-    from airflow.serialization.enums import DagAttributeTypes
 
 
 def _default_parent_group() -> TaskGroup | None:
@@ -491,15 +490,15 @@ def get_child_by_label(self, label: str) -> DAGNode:
         """Get a child task/TaskGroup by its label (i.e. task_id/group_id without the group_id prefix)."""
         return self.children[self.child_id(label)]
 
-    def serialize_for_task_group(self) -> tuple[DagAttributeTypes, Any]:
+    def serialize_for_task_group(self) -> list:
         """Serialize task group; required by DagNode."""
         from airflow.serialization.enums import DagAttributeTypes
         from airflow.serialization.serialized_objects import TaskGroupSerialization
 
-        return (
-            DagAttributeTypes.TASK_GROUP,
+        return [
+            DagAttributeTypes.TASK_GROUP.value,
             TaskGroupSerialization.serialize_task_group(self),
-        )
+        ]
 
     def hierarchical_alphabetical_sort(self):
         """