Accept kubeconfig as a dict (#361)

kr8s/_api.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import contextlib
+import copy
 import json
 import ssl
 import threading
@@ -64,7 +65,8 @@ def __init__(self, **kwargs) -> None:
         thread_loop_id = f"{thread_id}.{loop_id}"
         if thread_loop_id not in Api._instances:
             Api._instances[thread_loop_id] = weakref.WeakValueDictionary()
-        Api._instances[thread_loop_id][frozenset(kwargs.items())] = self
+        key = hash_kwargs(kwargs)
+        Api._instances[thread_loop_id][key] = self
 
     def __await__(self):
         async def f():
@@ -508,3 +510,11 @@ def namespace(self) -> str:
     @namespace.setter
     def namespace(self, value):
         self.auth.namespace = value
+
+
+def hash_kwargs(kwargs: dict):
+    key_kwargs = copy.copy(kwargs)
+    for key in key_kwargs:
+        if isinstance(key_kwargs[key], dict):
+            key_kwargs[key] = json.dumps(key_kwargs[key])
+    return frozenset(key_kwargs.items())

kr8s/_auth.py

@@ -4,6 +4,7 @@
 import binascii
 import json
 import os
+import pathlib
 import ssl
 
 import anyio
@@ -42,7 +43,7 @@ def __init__(
             if serviceaccount is not None
             else "/var/run/secrets/kubernetes.io/serviceaccount"
         )
-        self._kubeconfig_path = kubeconfig or os.environ.get(
+        self._kubeconfig_path_or_dict = kubeconfig or os.environ.get(
             "KUBECONFIG", "~/.kube/config"
         )
         self.__auth_lock = anyio.Lock()
@@ -60,7 +61,7 @@ async def reauthenticate(self) -> None:
             if self._url:
                 self.server = self._url
             else:
-                if self._kubeconfig_path is not False:
+                if self._kubeconfig_path_or_dict is not False:
                     await self._load_kubeconfig()
                 if self._serviceaccount and not self.server:
                     await self._load_service_account()
@@ -91,10 +92,19 @@ async def ssl_context(self):
 
     async def _load_kubeconfig(self) -> None:
         """Load kubernetes auth from kubeconfig."""
-        self._kubeconfig_path = os.path.expanduser(self._kubeconfig_path)
-        if not os.path.exists(self._kubeconfig_path):
-            return
-        self.kubeconfig = await KubeConfigSet(*self._kubeconfig_path.split(":"))
+        if isinstance(self._kubeconfig_path_or_dict, str) or isinstance(
+            self._kubeconfig_path_or_dict, pathlib.Path
+        ):
+            self._kubeconfig_path_or_dict = os.path.expanduser(
+                self._kubeconfig_path_or_dict
+            )
+            if not os.path.exists(self._kubeconfig_path_or_dict):
+                return
+            self.kubeconfig = await KubeConfigSet(
+                *self._kubeconfig_path_or_dict.split(":")
+            )
+        else:
+            self.kubeconfig = await KubeConfigSet(self._kubeconfig_path_or_dict)
         if self._use_context:
             try:
                 self._context = self.kubeconfig.get_context(self._use_context)

kr8s/_config.py

@@ -1,5 +1,6 @@
 # SPDX-FileCopyrightText: Copyright (c) 2024, Kr8s Developers (See LICENSE for list)
 # SPDX-License-Identifier: BSD 3-Clause License
+import pathlib
 from typing import Dict, List, Union
 
 import anyio
@@ -19,7 +20,9 @@
 
 class KubeConfigSet(object):
     def __init__(self, *paths_or_dicts: Union[List[str], List[Dict]]):
-        if isinstance(paths_or_dicts[0], str):
+        if isinstance(paths_or_dicts[0], str) or isinstance(
+            paths_or_dicts[0], pathlib.Path
+        ):
             self._configs = [KubeConfig(path) for path in paths_or_dicts]
         else:
             self._configs = [KubeConfig(config) for config in paths_or_dicts]
@@ -177,7 +180,7 @@ class KubeConfig(object):
     def __init__(self, path_or_config: Union[str, Dict]):
         self.path = None
         self._raw = None
-        if isinstance(path_or_config, str):
+        if isinstance(path_or_config, str) or isinstance(path_or_config, pathlib.Path):
             self.path = path_or_config
         else:
             self._raw = path_or_config

kr8s/asyncio/_api.py

@@ -4,6 +4,7 @@
 import threading
 
 from kr8s._api import Api as _AsyncApi
+from kr8s._api import hash_kwargs
 
 
 async def api(
@@ -52,7 +53,7 @@ async def api(
         _cls = _SyncApi
 
     async def _f(**kwargs):
-        key = frozenset(kwargs.items())
+        key = hash_kwargs(kwargs)
         thread_id = threading.get_ident()
         try:
             loop_id = id(asyncio.get_running_loop())

kr8s/tests/test_auth.py

@@ -100,6 +100,22 @@ async def test_kubeconfig(k8s_cluster):
     assert await api.whoami() == "kubernetes-admin"
 
 
+async def test_kubeconfig_dict(k8s_cluster):
+    config = yaml.safe_load(k8s_cluster.kubeconfig_path.read_text())
+    assert isinstance(config, dict)
+    api = await kr8s.asyncio.api(kubeconfig=config)
+    assert await api.get("pods", namespace=kr8s.ALL)
+    assert await api.whoami() == "kubernetes-admin"
+
+
+def test_kubeconfig_dict_sync(k8s_cluster):
+    config = yaml.safe_load(k8s_cluster.kubeconfig_path.read_text())
+    assert isinstance(config, dict)
+    api = kr8s.api(kubeconfig=config)
+    assert api.get("pods", namespace=kr8s.ALL)
+    assert api.whoami() == "kubernetes-admin"
+
+
 async def test_kubeconfig_context(kubeconfig_with_second_context):
     kubeconfig_path, context_name = kubeconfig_with_second_context
     api = await kr8s.asyncio.api(kubeconfig=kubeconfig_path, context=context_name)