Fix pod_override serialization in DAG details and executor path (apache#65407)

Sanitize default_args["executor_config"]["pod_override"] in DAG
details responses without changing other default_args values,
and make V1Pod serialization force the Kubernetes import path
so executor config no longer falls back to stringification.

This fixes two related Kubernetes serialization bugs: one where
the DAG details API could fail when pod_override was present in
default_args, and another where V1Pod objects could be flattened
into strings before reaching the Kubernetes executor. The updated
tests cover both the API response behavior and the serializer
regression with real Kubernetes pod objects.

Author: ephraimbuddy

airflow-core/src/airflow/api_fastapi/core_api/datamodels/dags.py

@@ -33,6 +33,7 @@
     field_validator,
 )
 
+from airflow._shared.module_loading import qualname
 from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel, make_partial_model
 from airflow.api_fastapi.core_api.datamodels.dag_tags import DagTagResponse
 from airflow.api_fastapi.core_api.datamodels.dag_versions import DagVersionResponse
@@ -44,6 +45,11 @@
     from airflow.serialization.definitions.param import SerializedParamsDict
 
 
+def _is_response_safe_pod_override(value: Any) -> bool:
+    """Whether a pod_override value is already safe to preserve in the response."""
+    return value is None or isinstance(value, str | int | float | Mapping | list)
+
+
 @cache
 def _get_file_token_serializer() -> URLSafeSerializer:
     """
@@ -216,6 +222,37 @@ def get_doc_md(cls, doc_md: str | None) -> str | None:
             return None
         return inspect.cleandoc(doc_md)
 
+    @field_validator("default_args", mode="before")
+    @classmethod
+    def get_default_args(cls, default_args: Mapping | None) -> Mapping | None:
+        """
+        Sanitize default_args for the API response.
+
+        Targets the common case where ``executor_config["pod_override"]`` is a
+        Kubernetes ``V1Pod``: when the value is not a JSON primitive
+        (``None``/``str``/``int``/``float``) or a ``Mapping``/``list``, it is
+        rewritten to a fully-qualified type-name string so the response stays
+        valid JSON. The container check is shallow — a ``Mapping`` or ``list``
+        whose contents are themselves non-serializable (e.g. nested ``V1Pod``)
+        will still raise during response serialization, as will any other
+        non-JSON values elsewhere in ``default_args``.
+        """
+        if default_args is None:
+            return None
+        executor_config = default_args.get("executor_config")
+        if not (isinstance(executor_config, Mapping) and "pod_override" in executor_config):
+            return default_args
+
+        pod_override = executor_config["pod_override"]
+        if _is_response_safe_pod_override(pod_override):
+            return default_args
+
+        sanitized_executor_config = dict(executor_config)
+        sanitized_executor_config["pod_override"] = qualname(pod_override)
+        result = dict(default_args)
+        result["executor_config"] = sanitized_executor_config
+        return result
+
     @field_validator("params", mode="before")
     @classmethod
     def get_params(cls, params: SerializedParamsDict | None) -> dict | None:

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

@@ -501,7 +501,11 @@ def serialize(
             )
         elif isinstance(var, list):
             return [cls.serialize(v, strict=strict) for v in var]
-        elif var.__class__.__name__ == "V1Pod" and _has_kubernetes() and isinstance(var, k8s.V1Pod):
+        elif (
+            var.__class__.__name__ == "V1Pod"
+            and _has_kubernetes(attempt_import=True)
+            and isinstance(var, k8s.V1Pod)
+        ):
             json_pod = PodGenerator.serialize_pod(var)
             return cls._encode(json_pod, type_=DAT.POD)
         elif isinstance(var, OutletEventAccessors):

airflow-core/tests/unit/api_fastapi/core_api/datamodels/test_dags.py

@@ -16,9 +16,15 @@
 # under the License.
 from __future__ import annotations
 
+from datetime import datetime, timedelta, timezone
+
 import pytest
 
-from airflow.api_fastapi.core_api.datamodels.dags import DAGResponse
+from airflow._shared.module_loading import qualname
+from airflow.api_fastapi.core_api.datamodels.dags import (
+    DAGDetailsResponse,
+    DAGResponse,
+)
 from airflow.utils.types import DagRunType
 
 
@@ -58,6 +64,77 @@ def _make_dag_response(**overrides) -> DAGResponse:
     return DAGResponse.model_validate(defaults)
 
 
+class TestGetDefaultArgsValidator:
+    """Test the get_default_args field_validator on DAGDetailsResponse."""
+
+    def _call_validator(self, value):
+        """Invoke the classmethod validator directly."""
+        return DAGDetailsResponse.get_default_args(value)
+
+    def test_none_returns_none(self):
+        assert self._call_validator(None) is None
+
+    def test_plain_dict_is_preserved(self):
+        result = self._call_validator({"retries": 3, "depends_on_past": False})
+        assert result == {"retries": 3, "depends_on_past": False}
+
+    def test_timedelta_values_are_preserved(self):
+        td = timedelta(minutes=5)
+        result = self._call_validator({"retry_delay": td})
+        assert result == {"retry_delay": td}
+
+    def test_datetime_values_are_preserved(self):
+        start_date = datetime(2024, 1, 1, tzinfo=timezone.utc)
+        result = self._call_validator({"start_date": start_date})
+        assert result == {"start_date": start_date}
+
+    def test_pod_override_is_replaced_with_type_name(self):
+        k8s = pytest.importorskip("kubernetes.client.models")
+        pod = k8s.V1Pod(metadata=k8s.V1ObjectMeta(name="test-pod"))
+        result = self._call_validator({"executor_config": {"pod_override": pod, "namespace": "custom"}})
+        assert result == {"executor_config": {"pod_override": qualname(pod), "namespace": "custom"}}
+
+    @pytest.mark.parametrize(
+        "pod_override",
+        [
+            pytest.param(None, id="none"),
+            pytest.param("already-serialized", id="string"),
+            pytest.param({"metadata": {"name": "pod"}}, id="dict"),
+            pytest.param([{"metadata": {"name": "pod"}}], id="list"),
+        ],
+    )
+    def test_serialized_pod_override_values_are_preserved(self, pod_override):
+        result = self._call_validator({"executor_config": {"pod_override": pod_override}})
+        assert result == {"executor_config": {"pod_override": pod_override}}
+
+    def test_serialized_pod_override_preserves_other_executor_config_keys(self):
+        executor_config = {
+            "pod_override": {"metadata": {"name": "pod"}},
+            "KubernetesExecutor": {"image": "custom-image"},
+        }
+
+        result = self._call_validator({"executor_config": executor_config})
+
+        assert result == {"executor_config": executor_config}
+
+    def test_non_serialized_pod_override_object_is_replaced_with_type_name(self):
+        class Opaque:
+            pass
+
+        value = Opaque()
+        result = self._call_validator({"executor_config": {"pod_override": value}})
+        assert result == {"executor_config": {"pod_override": qualname(value)}}
+
+    def test_non_pod_override_objects_are_left_unchanged(self):
+        class Opaque:
+            def to_dict(self):
+                return {"password": "secret"}
+
+        value = Opaque()
+        result = self._call_validator({"connection": value})
+        assert result["connection"] is value
+
+
 class TestIsBackfillable:
     @pytest.mark.parametrize(
         "timetable_periodic",

airflow-core/tests/unit/serialization/test_serialized_objects.py

@@ -85,6 +85,7 @@
 from airflow.sdk.definitions.param import Param
 from airflow.sdk.definitions.taskgroup import TaskGroup
 from airflow.sdk.execution_time.context import OutletEventAccessor, OutletEventAccessors
+from airflow.serialization import serialized_objects
 from airflow.serialization.definitions.assets import (
     SerializedAsset,
     SerializedAssetAlias,
@@ -1038,6 +1039,28 @@ def test_has_kubernetes_uses_existing_import(self):
 
         assert result is True
 
+    def test_serialize_v1pod_attempts_import_before_serializing(self, monkeypatch):
+        """Regression test: V1Pod serialization must call _has_kubernetes(attempt_import=True)."""
+        k8s = pytest.importorskip("kubernetes.client.models")
+        from airflow.providers.cncf.kubernetes.pod_generator import PodGenerator
+
+        calls = []
+
+        def fake_has_kubernetes(*, attempt_import=False):
+            calls.append(attempt_import)
+            return True
+
+        monkeypatch.setattr(serialized_objects, "_has_kubernetes", fake_has_kubernetes)
+        monkeypatch.setattr(serialized_objects, "k8s", k8s, raising=False)
+        monkeypatch.setattr(serialized_objects, "PodGenerator", PodGenerator, raising=False)
+
+        pod = k8s.V1Pod(metadata=k8s.V1ObjectMeta(name="test-pod"))
+        result = BaseSerialization.serialize(pod)
+
+        assert isinstance(result, dict), "V1Pod should serialize to a dict, not a string"
+        assert result.get(Encoding.TYPE) == DAT.POD, "V1Pod should have type DAT.POD"
+        assert True in calls
+
 
 @pytest.mark.db_test
 def test_serialized_dag_getitem_returns_task(dag_maker):