Refactor: Simplify code in providers/cncf

airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py

@@ -371,7 +371,7 @@ def sync(self) -> None:
         from airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils import ResourceVersion
 
         resource_instance = ResourceVersion()
-        for ns in resource_instance.resource_version.keys():
+        for ns in resource_instance.resource_version:
             resource_instance.resource_version[ns] = (
                 last_resource_version[ns] or resource_instance.resource_version[ns]
             )

airflow/providers/cncf/kubernetes/pod_generator.py

@@ -606,7 +606,7 @@ def merge_objects(base_obj, client_obj):
         base_obj_cp.update(client_obj_cp)
         return base_obj_cp
 
-    for base_key in base_obj.to_dict().keys():
+    for base_key in base_obj.to_dict():
         base_val = getattr(base_obj, base_key, None)
         if not getattr(client_obj, base_key, None) and base_val:
             if not isinstance(client_obj_cp, dict):

airflow/providers/cncf/kubernetes/triggers/pod.py

@@ -246,7 +246,7 @@ def define_container_state(self, pod: V1Pod) -> ContainerState:
         if pod_containers is None:
             return ContainerState.UNDEFINED
 
-        container = [c for c in pod_containers if c.name == self.base_container_name][0]
+        container = next(c for c in pod_containers if c.name == self.base_container_name)
 
         for state in (ContainerState.RUNNING, ContainerState.WAITING, ContainerState.TERMINATED):
             state_obj = getattr(container.state, state)

airflow/providers/cncf/kubernetes/utils/pod_manager.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import enum
+import itertools as it
 import json
 import logging
 import math
@@ -26,7 +27,7 @@
 from collections.abc import Iterable
 from contextlib import closing, suppress
 from dataclasses import dataclass
-from datetime import datetime, timedelta
+from datetime import timedelta
 from typing import TYPE_CHECKING, Generator, cast
 
 import pendulum
@@ -324,14 +325,13 @@ def await_pod_start(self, pod: V1Pod, startup_timeout: int = 120) -> None:
             (if pod is pending for too long, fails task)
         :return:
         """
-        curr_time = datetime.now()
+        curr_time = time.time()
         while True:
             remote_pod = self.read_pod(pod)
             if remote_pod.status.phase != PodPhase.PENDING:
                 break
             self.log.warning("Pod not yet started: %s", pod.metadata.name)
-            delta = datetime.now() - curr_time
-            if delta.total_seconds() >= startup_timeout:
+            if time.time() - curr_time >= startup_timeout:
                 msg = (
                     f"Pod took longer than {startup_timeout} seconds to start. "
                     "Check the pod events in kubernetes to determine why."
@@ -628,14 +628,12 @@ def read_pod(self, pod: V1Pod) -> V1Pod:
 
     def await_xcom_sidecar_container_start(self, pod: V1Pod) -> None:
         self.log.info("Checking if xcom sidecar container is started.")
-        warned = False
-        while True:
+        for attempt in it.count():
             if self.container_is_running(pod, PodDefaults.SIDECAR_CONTAINER_NAME):
                 self.log.info("The xcom sidecar container is started.")
                 break
-            if not warned:
+            if not attempt:
                 self.log.warning("The xcom sidecar container is not yet started.")
-                warned = True
             time.sleep(1)
 
     def extract_xcom(self, pod: V1Pod) -> str: