Handle non iterables more carefully while sorting serdag (apache#47095)

Author: amoghrajesh

airflow/models/serialized_dag.py

@@ -21,6 +21,7 @@
 
 import logging
 import zlib
+from collections.abc import Iterable
 from datetime import timedelta
 from typing import TYPE_CHECKING, Any
 
@@ -151,7 +152,10 @@ def _sort_serialized_dag_dict(cls, serialized_dag: Any):
             return {k: cls._sort_serialized_dag_dict(v) for k, v in sorted(serialized_dag.items())}
         elif isinstance(serialized_dag, list):
             if all(isinstance(i, dict) for i in serialized_dag):
-                if all("task_id" in i.get("__var", {}) for i in serialized_dag):
+                if all(
+                    isinstance(i.get("__var", {}), Iterable) and "task_id" in i.get("__var", {})
+                    for i in serialized_dag
+                ):
                     return sorted(
                         [cls._sort_serialized_dag_dict(i) for i in serialized_dag],
                         key=lambda x: x["__var"]["task_id"],

tests/models/test_serialized_dag.py

@@ -357,3 +357,17 @@ def test_new_dag_versions_are_created_if_there_is_a_dagrun(self, dag_maker, sess
 
         assert session.query(DagVersion).count() == 2
         assert session.query(SDM).count() == 2
+
+    def test_example_dag_sorting_serialised_dag(self, session):
+        """
+        This test asserts if different dag ids -- simple or complex, can be sorted
+        """
+
+        example_dags = self._write_example_dags()
+
+        for _, dag in example_dags.items():
+            # flip the tags, the sorting function should sort it alphabetically
+            if dag.tags:
+                dag.tags = sorted(dag.tags, reverse=True)
+            sorted_dag = SDM._sort_serialized_dag_dict(dag)
+            assert sorted_dag == dag