Add `pod_template_dict` field to `KubernetesPodOperator`
diff --git a/airflow/providers/cncf/kubernetes/operators/pod.py b/airflow/providers/cncf/kubernetes/operators/pod.py
index 55fe1e4c3668c..5153b85965be4 100644
--- a/airflow/providers/cncf/kubernetes/operators/pod.py
+++ b/airflow/providers/cncf/kubernetes/operators/pod.py
@@ -218,6 +218,7 @@ class KubernetesPodOperator(BaseOperator):
         /airflow/xcom/return.json in the container will also be pushed to an
         XCom when the container completes.
     :param pod_template_file: path to pod template file (templated)
+    :param pod_template_dict: pod template dictionary (templated)
     :param priority_class_name: priority class name for the launched Pod
     :param pod_runtime_info_envs: (Optional) A list of environment variables,
         to be set in the container.
@@ -267,6 +268,7 @@ class KubernetesPodOperator(BaseOperator):
         "labels",
         "config_file",
         "pod_template_file",
+        "pod_template_dict",
         "namespace",
         "container_resources",
         "volumes",
@@ -322,6 +324,7 @@ def __init__(
         log_events_on_failure: bool = False,
         do_xcom_push: bool = False,
         pod_template_file: str | None = None,
+        pod_template_dict: dict | None = None,
         priority_class_name: str | None = None,
         pod_runtime_info_envs: list[k8s.V1EnvVar] | None = None,
         termination_grace_period: int | None = None,
@@ -404,6 +407,7 @@ def __init__(
         self.log_events_on_failure = log_events_on_failure
         self.priority_class_name = priority_class_name
         self.pod_template_file = pod_template_file
+        self.pod_template_dict = pod_template_dict
         self.name = self._set_name(name)
         self.random_name_suffix = random_name_suffix
         self.termination_grace_period = termination_grace_period
@@ -897,6 +901,11 @@ def build_pod_request_obj(self, context: Context | None = None) -> k8s.V1Pod:
             pod_template = pod_generator.PodGenerator.deserialize_model_file(self.pod_template_file)
             if self.full_pod_spec:
                 pod_template = PodGenerator.reconcile_pods(pod_template, self.full_pod_spec)
+        elif self.pod_template_dict:
+            self.log.debug("Pod template dict found, will parse for base pod")
+            pod_template = pod_generator.PodGenerator.deserialize_model_dict(self.pod_template_dict)
+            if self.full_pod_spec:
+                pod_template = PodGenerator.reconcile_pods(pod_template, self.full_pod_spec)
         elif self.full_pod_spec:
             pod_template = self.full_pod_spec
         else:
diff --git a/tests/providers/cncf/kubernetes/operators/test_pod.py b/tests/providers/cncf/kubernetes/operators/test_pod.py
index 5e9cbbb9168a0..8402f0f6b2399 100644
--- a/tests/providers/cncf/kubernetes/operators/test_pod.py
+++ b/tests/providers/cncf/kubernetes/operators/test_pod.py
@@ -30,6 +30,7 @@
 from airflow.exceptions import AirflowException, AirflowSkipException, TaskDeferred
 from airflow.models import DAG, DagModel, DagRun, TaskInstance
 from airflow.models.xcom import XCom
+from airflow.providers.cncf.kubernetes import pod_generator
 from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator, _optionally_suppress
 from airflow.providers.cncf.kubernetes.secret import Secret
 from airflow.providers.cncf.kubernetes.triggers.pod import KubernetesPodTrigger
@@ -1019,6 +1020,64 @@ def test_pod_template_file_kwargs_override(self, randomize_name, pod_template_fi
             "run_id": "test",
         }
 
+    @pytest.mark.parametrize(("randomize_name",), ([True], [False]))
+    def test_pod_template_dict(self, randomize_name):
+        templated_pod = k8s.V1Pod(
+            metadata=k8s.V1ObjectMeta(
+                namespace="templatenamespace",
+                name="hello",
+                labels={"release": "stable"},
+            ),
+            spec=k8s.V1PodSpec(
+                containers=[],
+                init_containers=[
+                    k8s.V1Container(
+                        name="git-clone",
+                        image="registry.k8s.io/git-sync:v3.1.1",
+                        args=[
+                            "--repo=git@github.com:airflow/some_repo.git",
+                            "--branch={{ params.get('repo_branch', 'master') }}",
+                        ],
+                    ),
+                ],
+            ),
+        )
+        k = KubernetesPodOperator(
+            task_id="task",
+            random_name_suffix=randomize_name,
+            pod_template_dict=pod_generator.PodGenerator.serialize_pod(templated_pod),
+            labels={"hello": "world"},
+        )
+
+        # render templated fields before checking generated pod spec
+        k.render_template_fields(context={"params": {"repo_branch": "test_branch"}})
+        pod = k.build_pod_request_obj(create_context(k))
+
+        if randomize_name:
+            assert pod.metadata.name.startswith("hello")
+            assert pod.metadata.name != "hello"
+        else:
+            assert pod.metadata.name == "hello"
+
+        assert pod.metadata.labels == {
+            "hello": "world",
+            "release": "stable",
+            "dag_id": "dag",
+            "kubernetes_pod_operator": "True",
+            "task_id": "task",
+            "try_number": "1",
+            "airflow_version": mock.ANY,
+            "airflow_kpo_in_cluster": str(k.hook.is_in_cluster),
+            "run_id": "test",
+        }
+
+        assert pod.spec.init_containers[0].name == "git-clone"
+        assert pod.spec.init_containers[0].image == "registry.k8s.io/git-sync:v3.1.1"
+        assert pod.spec.init_containers[0].args == [
+            "--repo=git@github.com:airflow/some_repo.git",
+            "--branch=test_branch",
+        ]
+
     @patch(f"{POD_MANAGER_CLASS}.fetch_container_logs")
     @patch(f"{POD_MANAGER_CLASS}.await_container_completion", new=MagicMock)
     def test_no_handle_failure_on_success(self, fetch_container_mock):

I would like to propose adding a new templated field to kubernetes pod operator. Currently, operator has `pod_template_file` field to create base pod spec but you need to create a file to use this feature. IMHO it might be useful to accept stream or content, thus users might easily configure the base pod spec.

**Use Case**

I would like to write airflow dag which clones git repository and mounts volume for cloned branch. Branch name should be parameterized so that users can specify it while running the DAG. Example code:

```python
templated_pod = k8s.V1Pod(
      metadata=k8s.V1ObjectMeta(),
      spec=k8s.V1PodSpec(
          containers=[],
          init_containers=[
            k8s.V1Container(
              name='git-clone',
              image='registry.k8s.io/git-sync:v3.1.1',
              args=[
                  f'--repo=git@github.com:airflow/some_repo.git',
                  f'--branch={{ params.get("repo_branch", "master") }}',
                  '--root=/tmp/git',
                  '--dest=gitclone',
                  '--ssh=true',
                  '--wait=120',
                  '--one-time=true',
              ],
              volume_mounts=[self.git_sync_mount, self.ssh_key_mount],
              security_context=k8s.V1SecurityContext(run_as_user=65533),
              resources=k8s.V1ResourceRequirements(
                  requests={
                      "cpu": "300m",
                      "memory": "512Mi",
                  },
                  limits={
                      "cpu": "300m",
                      "memory": "512Mi",
                  },
              ),
          
          ],
          volumes=[],
      ),
)
serialized_pod = yaml.dump(
       pod_generator.PodGenerator.serialize_pod(templated_pod)
)
operator = KubernetesPodOperator(pod_template_content=serialized_pod, ...)

```

