Allow setting the context to use in the kr8s API client (#162)

docs/authentication.md

@@ -59,6 +59,17 @@ import kr8s
 client = kr8s.api(kubeconfig="/path/to/kube/config")
 ```
 
+#### Context
+
+If you have multiple contexts in your config and you do not want to use the default or currently selected one you can set this explicitly.
+
+```python
+import kr8s
+
+client = kr8s.api(context="foo-context")
+```
+
+
 ### Service Account
 
 When running inside a Pod with a service account, credentials will be mounted into `/var/run/secrets/kubernetes.io/serviceaccount` so `kr8s` will also check there. However you can specify an alternate path if you know that service account style credentials are stored elsewhere.

kr8s/_api.py

@@ -48,6 +48,7 @@ def __init__(self, **kwargs) -> None:
             kubeconfig=self._kubeconfig,
             serviceaccount=self._serviceaccount,
             namespace=kwargs.get("namespace"),
+            context=kwargs.get("context"),
         )
         thread_id = threading.get_ident()
         if thread_id not in Api._instances:

kr8s/_auth.py

@@ -20,6 +20,7 @@ def __init__(
         url=None,
         serviceaccount=None,
         namespace=None,
+        context=None,
     ) -> None:
         self.server = None
         self.client_cert_file = None
@@ -29,6 +30,8 @@ def __init__(
         self.username = None
         self.password = None
         self.namespace = namespace
+        self.active_context = None
+        self._use_context = context
         self._context = None
         self._cluster = None
         self._user = None
@@ -87,14 +90,26 @@ async def _load_kubeconfig(self) -> None:
             return
         async with await anyio.open_file(self._kubeconfig) as f:
             config = yaml.safe_load(await f.read())
-        if "current-context" in config:
+        if self._use_context:
+            try:
+                [self._context] = [
+                    c["context"]
+                    for c in config["contexts"]
+                    if c["name"] == self._use_context
+                ]
+                self.active_context = self._use_context
+            except ValueError as e:
+                raise ValueError(f"No such context {self._use_context}") from e
+        elif "current-context" in config:
             [self._context] = [
                 c["context"]
                 for c in config["contexts"]
                 if c["name"] == config["current-context"]
             ]
+            self.active_context = config["current-context"]
         else:
-            self.context = config["contexts"][0]["context"]
+            self._context = config["contexts"][0]["context"]
+            self.active_context = config["contexts"][0]["name"]
 
         [self._cluster] = [
             c["cluster"]

kr8s/asyncio/_api.py

@@ -10,6 +10,7 @@ async def api(
     kubeconfig: str = None,
     serviceaccount: str = None,
     namespace: str = None,
+    context: str = None,
     _asyncio: bool = True,
 ) -> _AsyncApi:
     """Create a :class:`kr8s.Api` object for interacting with the Kubernetes API.
@@ -46,4 +47,5 @@ async def _f(**kwargs):
         kubeconfig=kubeconfig,
         serviceaccount=serviceaccount,
         namespace=namespace,
+        context=context,
     )

kr8s/tests/test_auth.py

@@ -49,12 +49,33 @@ async def kubeconfig_with_token(k8s_cluster, k8s_token):
         yield f.name
 
 
+@pytest.fixture
+async def kubeconfig_with_second_context(k8s_cluster):
+    # Open kubeconfig and extract the certificates
+    kubeconfig = yaml.safe_load(k8s_cluster.kubeconfig_path.read_text())
+    kubeconfig["contexts"].append(
+        {"context": kubeconfig["contexts"][0]["context"], "name": "foo-context"}
+    )
+    with tempfile.NamedTemporaryFile() as f:
+        f.write(yaml.safe_dump(kubeconfig).encode())
+        f.flush()
+        yield f.name, kubeconfig["contexts"][1]["name"]
+
+
 async def test_kubeconfig(k8s_cluster):
     kubernetes = await kr8s.asyncio.api(kubeconfig=k8s_cluster.kubeconfig_path)
     version = await kubernetes.version()
     assert "major" in version
 
 
+async def test_kubeconfig_context(kubeconfig_with_second_context):
+    kubeconfig_path, context_name = kubeconfig_with_second_context
+    client = await kr8s.asyncio.api(kubeconfig=kubeconfig_path, context=context_name)
+    assert client.auth.active_context == context_name
+    version = await client.version()
+    assert "major" in version
+
+
 async def test_default_service_account(k8s_cluster):
     kubernetes = await kr8s.asyncio.api(kubeconfig=k8s_cluster.kubeconfig_path)
     assert (