code: def _kubernetes_pod_operator_args(operator_args):
      from kubernetes import client
  
      from airflow.kubernetes.secret import Secret
  
      # Set dynamic env variables like run-id, task-id etc from here.
      secrets = [
          Secret("env", secret, secret)
          for secret in operator_args.get("secrets", [])
      ]
      args = operator_args
      args.update(
          {
              "secrets": secrets,
              # Question for (savin):
              # Default timeout in airflow is 120. I can remove `startup_timeout_seconds` for now. how should we expose it to the user?
          }
      )
      # We need to explicitly add the `client.V1EnvVar` over here because
      # `pod_runtime_info_envs` doesn't accept arguments in dictionary form and strictly
      # Requires objects of type `client.V1EnvVar`
      additional_env_vars = [
          client.V1EnvVar(
              name=k,
              value_from=client.V1EnvVarSource(
                  field_ref=client.V1ObjectFieldSelector(
                      field_path=str(v)
                  )
              ),
          )
          for k, v in {
              "METAFLOW_KUBERNETES_POD_NAMESPACE": "metadata.namespace",
              "METAFLOW_KUBERNETES_POD_NAME": "metadata.name",
              "METAFLOW_KUBERNETES_POD_ID": "metadata.uid",
              "METAFLOW_KUBERNETES_SERVICE_ACCOUNT_NAME": "spec.serviceAccountName",
              "METAFLOW_KUBERNETES_NODE_IP": "status.hostIP",
          }.items()
      ]
      args["pod_runtime_info_envs"] = additional_env_vars
  
      resources = args.get("resources")
      # KubernetesPodOperator version 4.2.0 renamed `resources` to
      # `container_resources` (https://github.com/apache/airflow/pull/24673) / (https://github.com/apache/airflow/commit/45f4290712f5f779e57034f81dbaab5d77d5de85)
      # This was done because `KubernetesPodOperator` didn't play nice with dynamic task mapping and they had to
      # deprecate the `resources` argument. Hence, the below code path checks for the version of `KubernetesPodOperator`
      # and then sets the argument. If the version < 4.2.0 then we set the argument as `resources`.
      # If it is > 4.2.0 then we set the argument as `container_resources`
      # The `resources` argument of `KubernetesPodOperator` is going to be deprecated soon in the future.
      # So we will only use it for `KubernetesPodOperator` version < 4.2.0
      # The `resources` argument will also not work for `foreach`s.
      provider_version = get_kubernetes_provider_version()
      k8s_op_ver = create_absolute_version_number(provider_version)
      if (
          k8s_op_ver is None
          or k8s_op_ver
          < create_absolute_version_number(
              KUBERNETES_PROVIDER_FOREACH_VERSION
          )
      ):
          # Since the provider version is less than `4.2.0` so we need to use the `resources` argument
          # We need to explicitly parse `resources`/`container_resources` to `k8s.V1ResourceRequirements`,
          # otherwise airflow tries to parse dictionaries to `airflow.providers.cncf.kubernetes.backcompat.pod.Resources`
          # object via `airflow.providers.cncf.kubernetes.backcompat.backward_compat_converts.convert_resources` function.
          # This fails many times since the dictionary structure it expects is not the same as
          # `client.V1ResourceRequirements`.
          args["resources"] = client.V1ResourceRequirements(
              requests=resources["requests"],
              limits=None
              if "limits" not in resources
              else resources["limits"],
          )
      else:  # since the provider version is greater than `4.2.0` so should use the `container_resources` argument
          args["container_resources"] = client.V1ResourceRequirements(
              requests=resources["requests"],
              limits=None
              if "limits" not in resources
              else resources["limits"],
          )
          del args["resources"]
  
      if operator_args.get("execution_timeout"):
          args["execution_timeout"] = timedelta(
              **operator_args.get(
                  "execution_timeout",
              )
          )
      if operator_args.get("retry_delay"):
          args["retry_delay"] = timedelta(
              **operator_args.get("retry_delay")
          )
      return args
  
imports: import hashlib
  import json
  import sys
  import platform
  from collections import defaultdict
  from datetime import datetime, timedelta
label_desc: Use image node_id 12 for calling a global function with example usage: _kubernetes_pod_operator_args(operator_args) and returns: args