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

Code before:
        return val
    if isinstance(val, str):
        if val.strip().lower() == "true":
            return True
        if val.strip().lower() == "false":
            return False
    return None


class AsyncKubernetesHook(KubernetesHook):
    """Hook to use Kubernetes SDK asynchronously."""

    def __init__(
        self, config_dict: dict | None = None, connection_extras: dict | None = None, *args, **kwargs
    ):
        super().__init__(*args, **kwargs)

        self.config_dict = config_dict
        self._extras: dict | None = connection_extras
        self._event_polling_fallback = False

    async def _load_config(self):
        """Return Kubernetes API session for use with requests."""
        in_cluster = self._coalesce_param(self.in_cluster, await self._get_field("in_cluster"))
        cluster_context = self._coalesce_param(self.cluster_context, await self._get_field("cluster_context"))
        kubeconfig_path = await self._get_field("kube_config_path")
        kubeconfig = await self._get_field("kube_config")
        num_selected_configuration = sum(
            1 for o in [in_cluster, kubeconfig, kubeconfig_path, self.config_dict] if o
        )

        async def api_client_from_kubeconfig_file(_kubeconfig_path: str | None):
            await async_config.load_kube_config(
                config_file=_kubeconfig_path,
                client_configuration=self.client_configuration,
                context=cluster_context,
            )
            return _TimeoutAsyncK8sApiClient()

        if num_selected_configuration > 1:
            raise AirflowException(
                "Invalid connection configuration. Options kube_config_path, "
                "kube_config, in_cluster are mutually exclusive. "
                "You can only use one option at a time."
            )

        if in_cluster:
            self.log.debug(LOADING_KUBE_CONFIG_FILE_RESOURCE.format("within a pod"))
            self._is_in_cluster = True
            async_config.load_incluster_config()
            return _TimeoutAsyncK8sApiClient()

Review comment (performance):
Instead of an internal flag, would it be easier to use the `@cache` decorator?

Code after:
        return val
    if isinstance(val, str):
        if val.strip().lower() == "true":
            return True
        if val.strip().lower() == "false":
            return False
    return None


class AsyncKubernetesHook(KubernetesHook):
    """Hook to use Kubernetes SDK asynchronously."""

    def __init__(
        self, config_dict: dict | None = None, connection_extras: dict | None = None, *args, **kwargs
    ):
        super().__init__(*args, **kwargs)

        self.config_dict = config_dict
        self._extras: dict | None = connection_extras
        self._event_polling_fallback = False
        self._config_loaded = False

    async def _load_config(self):
        """Load Kubernetes configuration once per hook instance."""
        if self._config_loaded:
            return

        in_cluster = self._coalesce_param(self.in_cluster, await self._get_field("in_cluster"))
        cluster_context = self._coalesce_param(self.cluster_context, await self._get_field("cluster_context"))
        kubeconfig_path = await self._get_field("kube_config_path")
        kubeconfig = await self._get_field("kube_config")

        num_selected_configuration = sum(
            1 for o in [in_cluster, kubeconfig, kubeconfig_path, self.config_dict] if o
        )

        if num_selected_configuration > 1:
            raise AirflowException(
                "Invalid connection configuration. Options kube_config_path, "
                "kube_config, in_cluster, and config_dict are mutually exclusive. "
                "You can only use one option at a time."
            )

        if in_cluster:
            self.log.debug(LOADING_KUBE_CONFIG_FILE_RESOURCE.format("within a pod"))
            async_config.load_incluster_config()
            self._is_in_cluster = True
            self._config_loaded = True
            return

        # If above block does not return, we are not in a cluster.