Repository: apache/airflow
File: providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py (Python)

Code before:
            ]
        most_recent_start_time = max(pod_start_times)
        return pod_start_times.index(most_recent_start_time)


class _optionally_suppress(AbstractContextManager):
    """
    Returns context manager that will swallow and log exceptions.

    By default swallows descendents of Exception, but you can provide other classes through
    the vararg ``exceptions``.

    Suppression behavior can be disabled with reraise=True.

    :meta private:
    """

    def __init__(self, *exceptions, reraise: bool = False) -> None:
        self._exceptions = exceptions or (Exception,)
        self.reraise = reraise
        self.exception = None

    def __enter__(self):
        return self

    def __exit__(self, exctype, excinst, exctb) -> bool:
        error = exctype is not None
        matching_error = error and issubclass(exctype, self._exceptions)
        if (error and not matching_error) or (matching_error and self.reraise):
            return False
        if matching_error:
            self.exception = excinst
            logger = logging.getLogger(__name__)
            logger.exception(excinst)
        return True


def _normalize_labels_dict(labels: dict) -> dict:
    """Return a copy of the labels dict with all None values replaced by empty strings."""
    return {k: ("" if v is None else v) for k, v in labels.items()}


Review comment (suggestion):
This code will raise an AttributeError if `pod.to_dict().get("status")` returns None, since you cannot call `.get("start_time")` on None. The code should use chained get() calls with a default dictionary to handle the case where status is None, for example: `pod.to_dict().get("status", {}).get("start_time")`.
```suggestion
            pod.to_dict().get("status", {}).get("start_time") for pod in pod_list
```

Code after:

    def process_duplicate_label_pods(self, pod_list: list[k8s.V1Pod]) -> k8s.V1Pod:
        """
        Patch or delete the existing pod with duplicate labels.

        This is to handle an edge case that can happen only if reattach_on_restart
        flag is False, and the previous run attempt has failed because the task
        process has been killed externally by the cluster or another process.

        If the task process is killed externally, it breaks the code execution and
        immediately exists the task. As a result the pod created in the previous attempt
        will not be properly deleted or patched by cleanup() method.

        Return the newly created pod to be used for the next run attempt.
        """
        new_pod = pod_list.pop(self._get_most_recent_pod_index(pod_list))
        old_pod = pod_list[0]
        self.patch_already_checked(old_pod, reraise=False)
        if self.on_finish_action == OnFinishAction.DELETE_POD:
            self.process_pod_deletion(old_pod)
        return new_pod

    def _get_most_recent_pod_index(self, pod_list: list[k8s.V1Pod]) -> int:
        """Loop through a list of V1Pod objects and get the index of the most recent one."""
        pod_start_times: list[datetime.datetime] = [
            pod.to_dict().get("status").get("start_time") for pod in pod_list
        ]
        if not all(pod_start_times):
            self.log.info(
                "Unable to determine most recent pod using start_time (some pods have not started yet). Falling back to creation_timestamp from pod metadata."
            )
            pod_start_times: list[datetime.datetime] = [  # type: ignore[no-redef]
                pod.to_dict()
                .get("metadata", {})
                .get("creation_timestamp", datetime.datetime.now(tz=datetime.timezone.utc))
                for pod in pod_list
            ]
        most_recent_start_time = max(pod_start_times)
        return pod_start_times.index(most_recent_start_time)


class _optionally_suppress(AbstractContextManager):
    """
    Returns context manager that will swallow and log exceptions.

    By default swallows descendents of Exception, but you can provide other classes through
    the vararg ``exceptions``.

    Suppression behavior can be disabled with reraise=True.

    :meta private: