Create new classes automatically in kr8s.get

[jacobtomlinson]

Which project are you requesting an enhancement for?

kr8s

What do you need?

In #198 I made a bunch of improvements around how we can specify resource names in kr8s.get.

import kr8s

# All of the following are equivalent
ingresses = kr8s.get("ing")
ingresses = kr8s.get("ingress")
ingresses = kr8s.get("ingress.networking.k8s.io")
ingresses = kr8s.get("ingress.v1.networking.k8s.io")

However, we still don't support getting objects that we don't have a base class representation for.

>>> kr8s.get("mutatingwebhookconfigurations.admissionregistration.k8s.io/v1")
...

File ~/Projects/kr8s-org/kr8s/kr8s/_objects.py:1267, in get_class(kind, version, _asyncio)
   1259         if (
   1260             not version
   1261             and "." in group
   1262             and cls_group == group.split(".", 1)[1]
   1263             and cls_version == group.split(".", 1)[0]
   1264         ):
   1265             return cls
-> 1267 raise KeyError(f"No object registered for {kind}{'.' + group if group else ''}")

KeyError: 'No object registered for mutatingwebhookconfigurations.admissionregistration.k8s.io'

However, if we first create a container class for it with the class factory we can do it.

import kr8s

MutatingWebhookConfiguration = kr8s.objects.new_class("mutatingwebhookconfiguration", "admissionregistration.k8s.io/v1", namespaced=False, asyncio=False)

mwcs = kr8s.get("mutatingwebhookconfigurations.admissionregistration.k8s.io/v1")

Alternatives

In the kr8s.objects.object_from_spec(...) method there is a kwarg called allow_unknown_type which if set to True will call new_class automatically if it is passed a spec that it isn't aware of.

We could do this in kr8s.get too so that we can get any API object that we haven't seen before. Given that kr8s is striving to provide a batteries included and close to kubectl experience we may want to make this dynamic classing the default behaviour and switch the kwarg to enable folks to explicitly opt out.

A couple of things to take into consideration:

• I had to explicitly set namespaced=False to make this work because I chose a non-namespaced resource. We would need to update new_class to look this up automatically via the Kubernetes API.
• I explicitly set asyncio=False because I wrote all the examples here with the sync API. We would potentially want to create both a sync and async class when calling new_class or have kr8s.objects.new_class create sync objects by default and kr8s.asyncio.objects.new_class continue to create async objects.

[jcmcken]

I know this isn't convenient quite yet, but is there a pattern to use to accomplish it? I tried to use some of the code you provided in this issue, but not having much luck. It works with the mutatingwebhookconfiguration snippet you provided, but when I try with a custom resource it doesn't work.

My specific use case is Istio, which is pretty commonly used. I tried VirtualService in networking.istio.io/v1beta1. Accessing this works via kubectl, but I get the "no object registered" when I try to create a new class and then retrieve it (literally just replacing the kind and API version in the snippet you gave above)

EDIT: Nevermind, I guess I can't just call kr8s.objects.new_class without assigning it to something. That seemed to be my issue

[jcmcken]

It looks like the class that you generate with new_class is always async? When I pass asyncio=False and try to run VirtualService.get('foo', namespace='bar') I get an exception telling me I need to await the result

When I do await the result, I get a TypeError. So yeah, not exactly clear how to work with kr8s and custom resources

[jacobtomlinson]

Glad you figured out how to do it manually.

Nevermind, I guess I can't just call kr8s.objects.new_class without assigning it to something. That seemed to be my issue

Yeah if you don't assign it then it will get garbage collected. Perhaps we should keep a reference internally in kr8s somewhere to avoid this happening. Would you mind opening a ticket to track this?

When I pass asyncio=False and try to run VirtualService.get('foo', namespace='bar') I get an exception telling me I need to await the result

This sounds like a bug. Would you mind opening a separate issue for this?