Generic Resources

Sometimes you may need to interact with resources installed in the cluster that are not provided by default with a kubernetes installation. You can still interact with such resources using a generic resource.

Interface

lightkube.generic_resource.create_global_resource(group: str, version: str, kind: str, plural: str, verbs=None) -> Type[GenericGlobalResource]

Create a new class representing a global resource with the provided specifications.

Parameters

• group str - API group of the resource. Example stable.example.com.
• version str - API group version. Example v1.
• kind str - Resource name. Example Job.
• plural str - Resource collection name. Example jobs.

returns Subclass of GenericGlobalResource.

Source code in src/lightkube/generic_resource.py

def create_global_resource(group: str, version: str, kind: str, plural: str, verbs=None) -> Type[GenericGlobalResource]:
    """Create a new class representing a global resource with the provided specifications.

    **Parameters**

    * **group** `str` - API group of the resource. Example `stable.example.com`.
    * **version** `str` - API group version. Example `v1`.
    * **kind** `str` - Resource name. Example `Job`.
    * **plural** `str` - Resource collection name. Example `jobs`.

    **returns**  Subclass of `GenericGlobalResource`.
    """
    return _create_resource(False, group, version, kind, plural, verbs=verbs)

lightkube.generic_resource.create_namespaced_resource(group: str, version: str, kind: str, plural: str, verbs=None) -> Type[GenericNamespacedResource]

Create a new class representing a namespaced resource with the provided specifications.

Parameters

• group str - API group of the resource. Example stable.example.com.
• version str - API group version. Example v1.
• kind str - Resource name. Example Job.
• plural str - Resource collection name. Example jobs.

returns Subclass of GenericNamespacedResource.

Source code in src/lightkube/generic_resource.py

def create_namespaced_resource(
    group: str, version: str, kind: str, plural: str, verbs=None
) -> Type[GenericNamespacedResource]:
    """Create a new class representing a namespaced resource with the provided specifications.

    **Parameters**

    * **group** `str` - API group of the resource. Example `stable.example.com`.
    * **version** `str` - API group version. Example `v1`.
    * **kind** `str` - Resource name. Example `Job`.
    * **plural** `str` - Resource collection name. Example `jobs`.

    **returns**  Subclass of `GenericNamespacedResource`.
    """
    return _create_resource(True, group, version, kind, plural, verbs=verbs)

Examples

from lightkube import Client
from lightkube.generic_resource import create_namespaced_resource

Job = create_namespaced_resource('stable.example.com', 'v1', 'Job', 'jobs')

client = Client()
job = client.get(Job, name="job1", namespace="my-namespace")

A generic resource is itself a subclass of dict so you can access the content as you would do with a dictionary:

print(job["path"]["to"]["something"])

For conveniency, default resources attributes apiVersion, metadata, kind and status can be accessed using the attribute notation:

print(job.kind)
print(job.metadata)

Specifically metadata is also decoded using the model models.meta_v1.ObjectMeta:

print(job.metadata.name)

Since it's a dictionary you can create a resource manually as follow:

job = Job(metadata={"name": "job2", "namespace": "my-namespace"}, spec=...)
client.create(job)

Note

Since generic resources are schemaless, more attention need to be given to what attributes are available or you will get an error from the server.

Subresources Status and Scale are also defined:

job = client.get(Job.Status, name="job1", namespace="my-namespace")

Note

Only some resources may support Scale.

Convenience Functions for Generic Resources

Some helper functions are also included to make using generic resources easier:

lightkube.generic_resource.get_generic_resource(version, kind)

Query generic resources already defined using one of the other methods described in this module or via codecs.load_all_yaml(..., create_resources_for_crds=True)

Parameters

• version str - Resource version including the API group. Example stable.example.com/v1
• kind str - Resource kind. Example: CronTab

returns class representing the generic resource or None if it's not found

Source code in src/lightkube/generic_resource.py

def get_generic_resource(version, kind):
    """Query generic resources already defined using one of the other methods described in this module or via
    `codecs.load_all_yaml(..., create_resources_for_crds=True)`

    **Parameters**

    * **version** `str` - Resource version including the API group. Example `stable.example.com/v1`
    * **kind** `str` - Resource kind. Example: `CronTab`

    **returns** class representing the generic resource or `None` if it's not found
    """
    resource = resource_registry.get(version, kind)
    if resource is None or not issubclass(resource, (GenericGlobalResource, GenericNamespacedResource)):
        return None
    return resource

lightkube.generic_resource.load_in_cluster_generic_resources(client: Client)

Loads all in-cluster CustomResourceDefinitions as generic resources.

Once loaded, generic resources can be obtained from generic_resource.get_generic_resource(), or used implicitly such as when using codecs.load_all_yaml().

Parameters

• client Client - Lightkube Client to use to load the CRDs.

Source code in src/lightkube/generic_resource.py

def load_in_cluster_generic_resources(client: Client):
    """Loads all in-cluster CustomResourceDefinitions as generic resources.

    Once loaded, generic resources can be obtained from `generic_resource.get_generic_resource()`,
    or used implicitly such as when using `codecs.load_all_yaml()`.

    **Parameters**

    * **client** `Client` - Lightkube Client to use to load the CRDs.
    """
    crds = client.list(apiextensions.CustomResourceDefinition)
    for crd in crds:
        create_resources_from_crd(crd)

lightkube.generic_resource.async_load_in_cluster_generic_resources(client: AsyncClient) async

Loads all in-cluster CustomResourceDefinitions as generic resources.

Once loaded, generic resources can be obtained from generic_resource.get_generic_resource(), or used implicitly such as when using codecs.load_all_yaml().

Parameters

• client AsyncClient - Lightkube AsyncClient to use to load the CRDs.

Source code in src/lightkube/generic_resource.py

async def async_load_in_cluster_generic_resources(client: AsyncClient):
    """Loads all in-cluster CustomResourceDefinitions as generic resources.

    Once loaded, generic resources can be obtained from `generic_resource.get_generic_resource()`,
    or used implicitly such as when using `codecs.load_all_yaml()`.

    **Parameters**

    * **client** `AsyncClient` - Lightkube AsyncClient to use to load the CRDs.
    """
    crds = client.list(apiextensions.CustomResourceDefinition)
    async for crd in crds:
        create_resources_from_crd(crd)

lightkube.generic_resource.create_resources_from_crd(crd: apiextensions.CustomResourceDefinition)

Creates a generic resource for each version in a CustomResourceDefinition.

Source code in src/lightkube/generic_resource.py

def create_resources_from_crd(crd: apiextensions.CustomResourceDefinition):
    """Creates a generic resource for each version in a CustomResourceDefinition."""
    if crd.spec.scope == "Namespaced":
        creator = create_namespaced_resource
    elif crd.spec.scope == "Cluster":
        creator = create_global_resource
    else:
        raise ValueError(f"Unexpected scope for resource.  Expected 'Namespaced' or 'Cluster', got {crd.spec.scope}")

    for version in crd.spec.versions:
        creator(**_crd_to_dict(crd, version.name))

load_in_cluster_generic_resources loads all CRDs in the cluster as generic resources, removing the need for explicitly defining each resource needed. This is especially helpful for scripting around YAML files that may use unknown custom resources. For example, using the Kubernetes example of the CronTab CRD:

crontab.yaml:

apiVersion: "stable.example.com/v1"
kind: CronTab
metadata:
  name: my-new-cron-object
spec:
  cronSpec: "* * * * */5"
  image: my-awesome-cron-image

from pathlib import Path

from lightkube import Client
from lightkube.codecs import load_all_yaml
from lightkube.generic_resource import load_in_cluster_generic_resources

# This fails with error message:
# lightkube.core.exceptions.LoadResourceError: No module named 'lightkube.resources.stable_example_com_v1'. If using a CRD, ensure you define a generic resource.
resources = load_all_yaml(Path("crontab.yaml").read_text())

client = Client()
load_in_cluster_generic_resources(client)

# Now we can load_all_yaml (and use those loaded resources, for example to create them in cluster)
resources = load_all_yaml(Path("crontab.yaml").read_text())

create_resource_from_crd creates generic resources for each version of a CustomResourceDefinition object. For example:

from lightkube.generic_resource import create_resources_from_crd
from lightkube.resources.apiextensions_v1 import CustomResourceDefinition
from lightkube.models.apiextensions_v1 import (
    CustomResourceDefinitionNames,
    CustomResourceDefinitionSpec,
    CustomResourceDefinitionVersion,
)
versions = ['v1alpha1', 'v1']

crd = CustomResourceDefinition(

    spec=CustomResourceDefinitionSpec(
        group='some.group',
        names=CustomResourceDefinitionNames(
            kind='somekind',
            plural='somekinds',
        ),
        scope='Namespaced',
        versions=[
            CustomResourceDefinitionVersion(
                name=version,
                served=True,
                storage=True,
            ) for version in versions
        ],
    )
)

create_resources_from_crd(crd)  # Creates two generic resources, one for each above version

# To demonstrate this worked:
from lightkube.generic_resource import _created_resources
print("Dict of custom resources that have been defined in Lightkube:")
print(_created_resources)