Implement apply

examples/kubectl-ng/kubectl_ng/_apply.py

@@ -0,0 +1,45 @@
+# SPDX-FileCopyrightText: Copyright (c) 2023, Dask Developers, NVIDIA
+# SPDX-License-Identifier: BSD 3-Clause License
+
+import json
+
+import typer
+from rich.console import Console
+
+import kr8s.asyncio
+from kr8s.asyncio.objects import objects_from_files
+
+console = Console()
+
+
+async def apply(
+    filename: str = typer.Option(
+        "",
+        "--filename",
+        "-f",
+        help="Filename, directory, or URL to files identifying the resource to wait on",
+    ),
+):
+    api = await kr8s.asyncio.api()
+    try:
+        objs = await objects_from_files(filename, api)
+    except Exception as e:
+        console.print(f"[red]Error loading objects from {filename}[/red]: {e}")
+        raise typer.Exit(1)
+    successful = True
+    for obj in objs:
+        if not obj.annotations:
+            obj.raw["metadata"]["annotations"] = {}
+        obj.raw["metadata"]["annotations"][
+            "kubectl.kubernetes.io/last-applied-configuration"
+        ] = json.dumps(obj.raw)
+        try:
+            status = await obj.apply()
+            # TODO detect if created, modified or unchanged
+            console.print(f"[green]{obj.singular}/{obj.name} {status}[/green]")
+        except Exception as e:
+            console.print(f"[red]{obj.singular}/{obj.name} failed[/red]: {e}")
+            successful = False
+            continue
+    if not successful:
+        raise typer.Exit(1)

examples/kubectl-ng/kubectl_ng/cli.py

@@ -6,6 +6,7 @@
 import typer
 
 from ._api_resources import api_resources
+from ._apply import apply
 from ._get import get
 from ._version import version
 from ._wait import wait
@@ -30,6 +31,7 @@ def register(app, func):
 register(app, get)
 register(app, version)
 register(app, wait)
+register(app, apply)
 
 
 def go():

kr8s/_api.py

@@ -14,6 +14,7 @@
 
 from ._auth import KubeAuth
 from ._data_utils import dict_to_selector
+from ._exceptions import ServerStatusError
 
 ALL = "all"
 
@@ -141,7 +142,10 @@ async def call_api(
                     await self._create_session()
                     continue
                 else:
-                    raise
+                    status = e.response.json()
+                    raise ServerStatusError(
+                        e.response.status_code, status["message"]
+                    ) from e
             except ssl.SSLCertVerificationError:
                 # In some rare edge cases the SSL verification fails, so we try again
                 # a few times before giving up.

kr8s/_data_utils.py

@@ -66,3 +66,28 @@ def dict_to_selector(selector_dict: Dict) -> str:
         A Kubernetes selector string.
     """
     return ",".join(f"{k}={v}" for k, v in selector_dict.items())
+
+
+def diff_nested_dicts(dict1: dict, dict2: dict) -> dict:
+    """Return the difference between two nested dictionaries.
+
+    Parameters
+    ----------
+    dict1 : dict
+        The first dictionary to compare.
+    dict2 : dict
+        The second dictionary to compare.
+
+    Returns
+    -------
+    dict
+        A dictionary containing the differences between the two input
+        dictionaries.
+    """
+    diff = {}
+    for key in dict1.keys() | dict2.keys():
+        if isinstance(dict1.get(key), dict) and isinstance(dict2.get(key), dict):
+            diff[key] = diff_nested_dicts(dict1[key], dict2[key])
+        elif dict1.get(key) != dict2.get(key):
+            diff[key] = dict2.get(key)
+    return diff

kr8s/_exceptions.py

@@ -4,3 +4,12 @@ class NotFoundError(Exception):
 
 class ConnectionClosedError(Exception):
     """A connection has been closed."""
+
+
+class ServerStatusError(Exception):
+    """The server returned an error status code."""
+
+    def __init__(self, status_code, message):
+        self.status_code = status_code
+        self.message = message
+        super().__init__(message)

kr8s/_objects.py

@@ -8,7 +8,7 @@
 import pathlib
 import re
 import time
-from typing import Any, Dict, List, Optional, Type, Union
+from typing import Any, Dict, List, Literal, Optional, Type, Union
 
 import anyio
 import httpx
@@ -19,8 +19,13 @@
 import kr8s
 import kr8s.asyncio
 from kr8s._api import Api
-from kr8s._data_utils import dict_to_selector, dot_to_nested_dict, list_dict_unpack
-from kr8s._exceptions import NotFoundError
+from kr8s._data_utils import (
+    dict_to_selector,
+    diff_nested_dicts,
+    dot_to_nested_dict,
+    list_dict_unpack,
+)
+from kr8s._exceptions import NotFoundError, ServerStatusError
 from kr8s.asyncio.portforward import PortForward as AsyncPortForward
 from kr8s.portforward import PortForward as SyncPortForward
 
@@ -227,7 +232,7 @@ async def _exists(self, ensure=False) -> bool:
             raise NotFoundError(f"Object {self.name} does not exist")
         return False
 
-    async def create(self) -> None:
+    async def _create(self) -> None:
         """Create this object in Kubernetes."""
         async with self.api.call_api(
             "POST",
@@ -238,6 +243,44 @@ async def create(self) -> None:
         ) as resp:
             self.raw = resp.json()
 
+    async def create(self) -> None:
+        """Create this object in Kubernetes."""
+        await self._create()
+
+    async def apply(self) -> Literal["created", "modified", "unchanged"]:
+        """Apply this object to Kubernetes."""
+        try:
+            existing = await self.get(self.name, self.namespace)
+            print(existing.raw)
+            print(self.raw)
+            if (
+                existing.spec.to_dict() == self.spec.to_dict()
+                and existing.annotations == self.annotations
+                and existing.labels == self.labels
+            ):
+                return "unchanged"
+            metadata = {
+                k: v for k, v in self.metadata.items() if k in ["labels", "annotations"]
+            }
+            existing_metadata = {
+                k: v
+                for k, v in existing.metadata.items()
+                if k in ["labels", "annotations"]
+            }
+            patch = diff_nested_dicts(
+                {"spec": existing.spec.to_dict(), "metadata": existing_metadata},
+                {"spec": self.spec.to_dict(), "metadata": metadata},
+            )
+            print(patch)
+            await self._patch(
+                patch,
+                force=True,
+            )
+            return "modified"
+        except NotFoundError:
+            await self._create()
+            return "created"
+
     async def delete(self, propagation_policy: str = None) -> None:
         """Delete this object from Kubernetes."""
         data = {}
@@ -280,7 +323,7 @@ async def patch(self, patch, *, subresource=None) -> None:
         """Patch this object in Kubernetes."""
         await self._patch(patch, subresource=subresource)
 
-    async def _patch(self, patch: Dict, *, subresource=None) -> None:
+    async def _patch(self, patch: Dict, *, subresource=None, **kwargs) -> None:
         """Patch this object in Kubernetes."""
         url = f"{self.endpoint}/{self.name}"
         if subresource:
@@ -299,6 +342,8 @@ async def _patch(self, patch: Dict, *, subresource=None) -> None:
             if e.response.status_code == 404:
                 raise NotFoundError(f"Object {self.name} does not exist") from e
             raise e
+        except ServerStatusError as e:
+            raise e
 
     async def scale(self, replicas: int = None) -> None:
         """Scale this object in Kubernetes."""

kr8s/tests/test_data_utils.py

@@ -1,7 +1,12 @@
 # SPDX-FileCopyrightText: Copyright (c) 2023, Dask Developers, NVIDIA
 # SPDX-License-Identifier: BSD 3-Clause License
 
-from kr8s._data_utils import dict_to_selector, dot_to_nested_dict, list_dict_unpack
+from kr8s._data_utils import (
+    dict_to_selector,
+    diff_nested_dicts,
+    dot_to_nested_dict,
+    list_dict_unpack,
+)
 
 
 def test_list_dict_unpack():
@@ -22,3 +27,18 @@ def test_dot_to_nested_dict():
 def test_dict_to_selector():
     assert dict_to_selector({"foo": "bar"}) == "foo=bar"
     assert dict_to_selector({"foo": "bar", "baz": "qux"}) == "foo=bar,baz=qux"
+
+
+def test_diff_nested_dicts():
+    assert diff_nested_dicts({"foo": "bar"}, {"foo": "bar"}) == {}
+    assert diff_nested_dicts({"foo": "bar"}, {"foo": "baz"}) == {"foo": "baz"}
+    assert diff_nested_dicts({"foo": "bar"}, {"foo": "bar", "baz": "qux"}) == {
+        "baz": "qux"
+    }
+    assert diff_nested_dicts({"foo": [{"bar": "baz"}]}, {"foo": [{"bar": "qux"}]}) == {
+        "foo": [{"bar": "qux"}]
+    }
+    assert diff_nested_dicts(
+        {"foo": [{"bar": "baz"}, {"fizz": "buzz"}]},
+        {"foo": [{"bar": "qux"}, {"fizz": "buzz"}]},
+    ) == {"foo": [{"bar": "qux"}, {"fizz": "buzz"}]}

kr8s/tests/test_objects.py

@@ -198,6 +198,21 @@ async def test_pod_metadata(example_pod_spec, ns):
     await pod.delete()
 
 
+async def test_pod_apply(example_pod_spec):
+    pod = await Pod(example_pod_spec)
+    await pod.apply()
+    assert "name" in pod.metadata
+    assert "hello" in pod.labels
+    assert "foo" not in pod.labels
+
+    pod.raw["metadata"]["labels"]["foo"] = "bar"
+    await pod.apply()
+    await pod.refresh()
+    assert "foo" in pod.labels
+
+    await pod.delete()
+
+
 async def test_pod_missing_labels_annotations(example_pod_spec):
     del example_pod_spec["metadata"]["labels"]
     del example_pod_spec["metadata"]["annotations"]