Load/Dump kubernetes objects

Convert models from/to dict

All lightkube models allow to convert from/to dicts using the methods .from_dict() and .to_dict(). For example you can create an ObjectMeta with

from lightkube.models.meta_v1 import ObjectMeta
meta = ObjectMeta.from_dict({'name': 'my-name', 'labels': {'key': 'value'}})

and transform it back with

meta_dict = meta.to_dict()

Dict representations can then be serialized/deserialized in JSON or YAML.

Load resource objects

It is possible to load dynamically a resource object using the function lightkube.codecs.from_dict()

from lightkube import codecs

obj = codecs.from_dict({
        'apiVersion': 'v1',
        'kind': 'ConfigMap',
        'metadata': {'name': 'config-name', 'labels': {'label1': 'value1'}},
        'data': {
            'file1.txt': 'some content here',
            'file2.txt': 'some other content'
        }
})
print(type(obj))

Output: <class 'lightkube.resources.core_v1.ConfigMap'>

Note

Only known resources can be loaded. These are either kubernetes standard resources or generic resources manually defined. You can register further resources using the resource_registry.

Load from YAML

Kubernetes resource defined in a YAML file can be easily loaded using the following function:

lightkube.codecs.load_all_yaml(stream: Union[str, TextIO], context: Optional[dict] = None, template_env=None, create_resources_for_crds: bool = False) -> List[AnyResource]

Load kubernetes resource objects defined as YAML. See from_dict regarding how resource types are detected. Returns a list of resource objects or raise a LoadResourceError. Skips any empty YAML documents in the stream, returning an empty list if all YAML documents are empty. Deep parse any items from .*List resources.

parameters

• stream - A file-like object or a string representing a yaml file or a template resulting in a yaml file.
• context - When is not None the stream is considered a jinja2 template and the context will be used during templating.
• template_env - jinja2 template environment to be used for templating. When absent a standard environment is used.
• create_resources_for_crds - If True, a generic resource will be created for every version of every CRD found that does not already have a generic resource. There will be no side effect for any CRD that already has a generic resource. Else if False, no generic resources will be created. Default is False.

NOTE: When using the template functionality (setting the context parameter), the dependency module jinja2 need to be installed.

Source code in src/lightkube/codecs.py

def load_all_yaml(
    stream: Union[str, TextIO],
    context: Optional[dict] = None,
    template_env=None,
    create_resources_for_crds: bool = False,
) -> List[AnyResource]:
    """Load kubernetes resource objects defined as YAML. See `from_dict` regarding how resource types are detected.
    Returns a list of resource objects or raise a `LoadResourceError`.  Skips any empty YAML documents in the
    stream, returning an empty list if all YAML documents are empty. Deep parse any items from .*List resources.

    **parameters**

    * **stream** - A file-like object or a string representing a yaml file or a template resulting in
        a yaml file.
    * **context** - When is not `None` the stream is considered a `jinja2` template and the `context`
        will be used during templating.
    * **template_env** - `jinja2` template environment to be used for templating. When absent a standard
        environment is used.
    * **create_resources_for_crds** - If True, a generic resource will be created for every version
        of every CRD found that does not already have a generic resource.  There will be no side
        effect for any CRD that already has a generic resource.  Else if False, no generic resources
         will be created.  Default is False.

    **NOTE**: When using the template functionality (setting the context parameter), the dependency
        module `jinja2` need to be installed.
    """
    if context is not None:
        stream = _template(stream, context=context, template_env=template_env)

    def _flatten(objects: Iterator) -> List[AnyResource]:
        """Flatten resources which have a kind = *List."""
        resources = []
        for obj in objects:
            if obj is None:
                continue
            if isinstance(obj, Mapping) and obj.get("kind", "").endswith("List"):
                resources += _flatten(obj.get("items") or [])
            else:
                res = from_dict(obj)
                resources.append(res)

                if create_resources_for_crds is True and res.kind == "CustomResourceDefinition":
                    create_resources_from_crd(res)
        return resources

    return _flatten(yaml.safe_load_all(stream))

Example

from lightkube import Client, codecs

client = Client()
with open('deployment.yaml') as f:
    for obj in codecs.load_all_yaml(f):
        client.create(obj)

Note

Only defined resources can be loaded. These are either kubernetes standard resources or generic resources manually defined.

If we have a YAML file that both defines a CRD and loads an instance of it, we can use create_resources_for_crds=True, like:

from lightkube import Client, codecs

client = Client()
with open('file-with-crd-and-instance.yaml') as f:
    for obj in codecs.load_all_yaml(f, create_resources_for_crds=True):
        client.create(obj)

This results in a generic resource being created for any CustomResourceDefinition in the YAML file.

It is also possible to create resources from a jinja2 template passing the parameter context.

For example assuming service.tmpl has the following content:

apiVersion: v1
kind: Service
metadata:
  name: nginx
  labels:
    run: my-nginx
    env: {{env}}
spec:
  type: NodePort
  ports:
  - port: 8080
    targetPort: 80
    protocol: TCP
  selector:
    run: my-nginx
    env: {{env}}

can be used as follow:

with open('service.tmpl') as f:
    # render the template using `context` and return the corresponding resource objects.
    objs = codecs.load_all_yaml(f, context={'env': 'prd'})
    print(objs[0].metadata.labels['env'])  # prints `prd`

Dump to YAML

The function lightkube.codecs.dump_all_yaml(...) can be used to dump resource objects as YAML.

lightkube.codecs.dump_all_yaml(resources: List[AnyResource], stream: Optional[TextIO] = None, indent: int = 2) -> Union[str, None]

Write kubernetes resource objects as YAML into an open file.

parameters

• resources - List of resources to write on the file
• stream - Path to a file where to write the resources. When not set the content is returned as a string.
• indent - Number of characters for indenting nasted blocks.

Source code in src/lightkube/codecs.py

def dump_all_yaml(resources: List[AnyResource], stream: Optional[TextIO] = None, indent: int = 2) -> Union[str, None]:
    """Write kubernetes resource objects as YAML into an open file.

    **parameters**

    * **resources** - List of resources to write on the file
    * **stream** - Path to a file where to write the resources. When not set the content is returned
      as a string.
    * **indent** - Number of characters for indenting nasted blocks.
    """
    res = [r.to_dict() for r in resources]
    return yaml.safe_dump_all(res, stream, indent=indent)

Example

from lightkube.resources.core_v1 import ConfigMap
from lightkube.models.meta_v1 import ObjectMeta
from lightkube import codecs

cm = ConfigMap(
    apiVersion='v1', kind='ConfigMap',
    metadata=ObjectMeta(name='xyz', labels={'x': 'y'})
)
with open('deployment-out.yaml', 'w') as fw:
    codecs.dump_all_yaml([cm], fw)

Sorting resource objects

Sometimes you have a manifest of resources where some depend on others. For example, consider, the following yaml_with_dependencies.yaml file:

kind: ClusterRoleBinding
roleRef:
  kind: ClusterRole
  name: example-cluster-role-binding
subjects:
  - kind: ServiceAccount
    name: example-service-account
...
---
kind: ClusterRole
metadata:
  name: example-cluster-role
...
---
kind: ServiceAccount
metadata:
  name: example-service-account

where we have a ClusterRoleBinding that uses a ClusterRole and ServiceAccount. In cases like this, the order in which we apply these resources matters as the ClusterRoleBinding depends on the others. To sort these objects so that we do not encounter API errors when applying them, use sort_objects(...).

lightkube.sort_objects(objs: List[r.Resource], by: str = 'kind', reverse: bool = False) -> List[r.Resource]

Sorts a list of resource objects by a sorting schema, returning a new list

parameters

• objs - list of resource objects to be sorted
• by - (optional) sorting schema. Possible values:
  • 'kind' - sorts by kind, ranking objects in an order that is suitable for batch-applying many resources. For example, Namespaces and ServiceAccounts are sorted ahead of ClusterRoleBindings or Pods that might use them. The reverse of this order is suitable for batch-deleting. See _kind_rank_function for full details on sorting
• reverse - (optional) if True, sorts in reverse order

Source code in src/lightkube/core/sort_objects.py

def sort_objects(objs: List[r.Resource], by: str = "kind", reverse: bool = False) -> List[r.Resource]:
    """Sorts a list of resource objects by a sorting schema, returning a new list

    **parameters**

    * **objs** - list of resource objects to be sorted
    * **by** - *(optional)* sorting schema. Possible values:
        * `'kind'` - sorts by kind, ranking objects in an order that is suitable for batch-applying
          many resources.  For example, Namespaces and ServiceAccounts are sorted ahead of
          ClusterRoleBindings or Pods that might use them.  The reverse of this order is suitable
          for batch-deleting.
          See _kind_rank_function for full details on sorting
    * **reverse** - *(optional)* if `True`, sorts in reverse order
    """
    if by == "kind":
        objs = sorted(objs, key=_kind_rank_function, reverse=reverse)
    else:
        raise ValueError(f"Unknown sorting schema: {by}")
    return objs

Revisiting the example above, we can apply from yaml_with_dependencies.yaml by:

from lightkube import Client, codecs, sort_objects

client = Client()
with open('yaml_with_dependencies.yaml') as f:
    objects = codecs.load_all_yaml(f)
    for obj in sort_objects(objects):
        client.create(obj)

sort_objects orders the objects in a way that is friendly to applying them as a batch, allowing us to loop through them as normal.

Similarly, problems can arise when deleting a batch of objects. For example, consider the manifest crs_and_crds.yaml:

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
...
spec:
  names:
    kind: SomeNewCr
    ...
---
kind: SomeNewCr
metadata:
  name: instance-of-new-cr

Deleting this in a loop like above would first delete the CustomResourceDefinition, resulting in all instances of SomeNewCr to be deleted implicitly. When we then attempted to delete instance-of-new-cr, we would encounter an API error.
Use codecs.sort_objects(..., reverse=True) to avoid this issue:

from lightkube import Client, codecs, sort_objects

client = Client()
with open('crs_amd_crds.yaml') as f:
    objects = codecs.load_all_yaml(f)
    for obj in sort_objects(objects, reverse=True):
        client.create(obj)

This orders the objects in a way that is friendly for deleting them as a batch.

Resource Registry

lightkube.codecs.resource_registry = ResourceRegistry() module-attribute

The singleton resource_registry allows to register a custom resource, so that it can be used by the load functions on this module:

from lightkube import codecs

codecs.resource_registry.register(MyCustomResource)

with open('service.yaml') as f:
    # Now `MyCustomResource` can be loaded
    objs = codecs.load_all_yaml(f)

register can also be used as a decorator:

from lightkube.core.resource import NamespacedResource
from lightkube.codecs import resource_registry

@resource_registry.register
class MyCustomResource(NamespacedResource):
    ...