Improve flexibility when getting resources by name (#198)

examples/kubectl-ng/kubectl_ng/_get.py

@@ -78,16 +78,7 @@ async def get(
     kubernetes = await kr8s.asyncio.api()
     if all_namespaces:
         namespace = kr8s.ALL
-    api_resources = await kubernetes.api_resources()
     for kind in resources:
-        for api_resource in api_resources:
-            if (
-                kind == api_resource["name"]
-                or kind == api_resource["singularName"]
-                or ("shortNames" in api_resource and kind in api_resource["shortNames"])
-            ):
-                kind = api_resource["name"]
-                break
         response = await kubernetes.get(
             kind,
             namespace=namespace,

kr8s/_api.py

@@ -247,11 +247,16 @@ async def _get_kind(
         if watch:
             params["watch"] = "true" if watch else "false"
             kwargs["stream"] = True
+        resources = await self._api_resources()
+        for resource in resources:
+            if "shortNames" in resource and kind in resource["shortNames"]:
+                kind = resource["name"]
+                break
         params = params or None
         obj_cls = get_class(kind, _asyncio=self._asyncio)
         async with self.call_api(
             method="GET",
-            url=kind,
+            url=obj_cls.endpoint,
             version=obj_cls.version,
             namespace=namespace if obj_cls.namespaced else None,
             params=params,

kr8s/_objects.py

@@ -1208,39 +1208,67 @@ def column_definitions(self) -> List[Dict]:
 
 
 def get_class(
-    kind: str, version: Optional[str] = None, _asyncio: bool = True
+    kind: str,
+    version: Optional[str] = None,
+    _asyncio: bool = True,
 ) -> Type[APIObject]:
     """Get an APIObject subclass by kind and version.
 
     Args:
         kind: The Kubernetes resource kind.
-        version: The Kubernetes API version.
+        version: The Kubernetes API group/version.
 
     Returns:
         An APIObject subclass.
 
     Raises:
         KeyError: If no object is registered for the given kind and version.
     """
+    group = None
+    if "/" in kind:
+        kind, version = kind.split("/", 1)
+    if "." in kind:
+        kind, group = kind.split(".", 1)
+    if version and "/" in version:
+        if group:
+            raise ValueError("Cannot specify group in both kind and version")
+        group, version = version.split("/", 1)
+    kind = kind.lower()
 
     def _walk_subclasses(cls):
         yield cls
         for subcls in cls.__subclasses__():
             yield from _walk_subclasses(subcls)
 
     for cls in _walk_subclasses(APIObject):
+        if not hasattr(cls, "version"):
+            continue
+        if "/" in cls.version:
+            cls_group, cls_version = cls.version.split("/")
+        else:
+            cls_group, cls_version = None, cls.version
         if (
             hasattr(cls, "kind")
-            and (cls.kind == kind or cls.singular == kind or cls.plural == kind)
-            and (version is None or cls.version == version)
             and cls._asyncio == _asyncio
+            and (cls.kind == kind or cls.singular == kind or cls.plural == kind)
         ):
-            return cls
-    raise KeyError(f"No object registered for {version}/{kind}")
+            if (group is None or cls_group == group) and (
+                version is None or cls_version == version
+            ):
+                return cls
+            if (
+                not version
+                and "." in group
+                and cls_group == group.split(".", 1)[1]
+                and cls_version == group.split(".", 1)[0]
+            ):
+                return cls
+
+    raise KeyError(f"No object registered for {kind}{'.' + group if group else ''}")
 
 
 def new_class(
-    kind: str, version: str = None, asyncio: bool = True, namespaced=True
+    kind: str, version: Optional[str] = None, asyncio: bool = True, namespaced=True
 ) -> Type[APIObject]:
     """Create a new APIObject subclass.
 
@@ -1253,6 +1281,8 @@ def new_class(
     Returns:
         A new APIObject subclass.
     """
+    if "." in kind:
+        kind, version = kind.split(".", 1)
     if version is None:
         version = "v1"
     return type(

kr8s/tests/test_api.py

@@ -247,3 +247,19 @@ async def get_api():
     asyncio.run(get_api())
     asyncio.set_event_loop(asyncio.new_event_loop())
     asyncio.run(get_api())
+
+
+async def test_api_names(example_pod_spec, ns):
+    pod = await Pod(example_pod_spec)
+    await pod.create()
+    assert pod in await kr8s.asyncio.get("pods", namespace=ns)
+    assert pod in await kr8s.asyncio.get("pods/v1", namespace=ns)
+    assert pod in await kr8s.asyncio.get("Pod", namespace=ns)
+    assert pod in await kr8s.asyncio.get("pod", namespace=ns)
+    assert pod in await kr8s.asyncio.get("po", namespace=ns)
+    await pod.delete()
+
+    await kr8s.asyncio.get("roles", namespace=ns)
+    await kr8s.asyncio.get("roles.rbac.authorization.k8s.io", namespace=ns)
+    await kr8s.asyncio.get("roles.v1.rbac.authorization.k8s.io", namespace=ns)
+    await kr8s.asyncio.get("roles.rbac.authorization.k8s.io/v1", namespace=ns)

kr8s/tests/test_objects.py

@@ -20,7 +20,7 @@
 )
 from kr8s.asyncio.portforward import PortForward
 from kr8s.objects import Pod as SyncPod
-from kr8s.objects import get_class, object_from_spec
+from kr8s.objects import get_class, new_class, object_from_spec
 
 DEFAULT_TIMEOUT = httpx.Timeout(30)
 CURRENT_DIR = pathlib.Path(__file__).parent
@@ -415,6 +415,15 @@ class MyResource(APIObject):
     get_class("MyResource", "foo.kr8s.org/v1alpha1")
 
 
+async def test_new_class_registration():
+    with pytest.raises(KeyError):
+        get_class("MyOtherResource", "foo.kr8s.org/v1alpha1")
+
+    MyOtherResource = new_class("MyOtherResource.foo.kr8s.org/v1alpha1")  # noqa: F841
+
+    get_class("MyOtherResource", "foo.kr8s.org/v1alpha1")
+
+
 async def test_deployment_scale(example_deployment_spec):
     deployment = await Deployment(example_deployment_spec)
     await deployment.create()