Lightweight Kubernetes Client
Lightkube is a modern, lightweight Python client library for Kubernetes, providing a simple, type-hinted interface for interacting with the Kubernetes API. It supports both synchronous and asynchronous operations, facilitates loading resources from YAML, and automatically handles pagination. The current version is 0.19.1 and the library maintains an active release cadence with frequent updates to support new Kubernetes versions and add features.
Common errors
-
lightkube.core.exceptions.ConfigError: Configuration file ~/.kube/config not found.
cause Lightkube could not find a valid Kubernetes configuration file (kubeconfig) in standard locations or was unable to load in-cluster configuration.fixEnsure a `~/.kube/config` file exists and is correctly configured, or that the application is running inside a Kubernetes cluster with a service account mounted. -
TypeError: 'ListIterable' object is not an iterator
cause Attempting to use `next()` directly on the result of `client.list()` after `lightkube v0.17.0`.fixBefore calling `next()`, explicitly convert the `ListIterable` to an iterator: `my_iterator = iter(client.list(...)); next(my_iterator)`. -
httpx.HTTPStatusError: Server error '500 Internal Server Error' for URL: ...
cause A generic HTTP error (e.g., 500, 404, 403) returned from the Kubernetes API server, wrapped by `lightkube.ApiError`.fixInspect the `e.status` attribute of the caught `lightkube.ApiError` for more details on the Kubernetes API error (e.g., `e.status.message`, `e.status.code`). Check API server logs for the specific issue. -
from lightkube.resources.apps_v1 import Deployment ImportError: cannot import name 'Deployment' from 'lightkube.resources.apps_v1'
cause This usually indicates an incompatible `lightkube-models` version, where the requested resource definition is not present or has a different path in the installed models, or that `lightkube-models` isn't installed for the specific K8s version.fixVerify that `lightkube-models` is installed and its version range is compatible with the target Kubernetes cluster API version. For example, for K8s 1.28, you might need `pip install 'lightkube-models>=1.28,<1.29'`.
Warnings
- breaking The `client.list()` method changed its return type from an `Iterator` to an `Iterable` in `v0.17.0`. If you were directly using `next()` on the result, you must now explicitly call `iter()` first (e.g., `next(iter(client.list(...)))`). Iterating with a `for` loop remains unchanged.
- breaking Since `kubernetes-models v1.33`, resource models will automatically have `apiVersion` and `kind` set upon initialization. Code that explicitly sets these fields on models might encounter unexpected behavior or redundancy.
- gotcha Lightkube's compatibility with `httpx` versions can be volatile. Specific `lightkube` versions might pin or limit `httpx` to avoid breaking changes in `httpx` itself. For instance, `v0.16.0` limited `httpx < 0.28.0` due to a breaking change in `httpx 0.28.0`.
- breaking Official Python version support has evolved. `lightkube v0.16.0` dropped official support for Python 3.8 while adding support for Python 3.13.
- gotcha The `lightkube-models` package needs to be installed in a version compatible with your Kubernetes cluster's API version. Mismatched `lightkube-models` versions can lead to missing fields or incorrect API interactions.
Install
-
pip install lightkube
Imports
- Client
from lightkube import Client
- AsyncClient
from lightkube import AsyncClient
- codecs
from lightkube import codecs
- Pod
from lightkube.models.core_v1 import Pod
from lightkube.resources.core_v1 import Pod
- ObjectMeta
from lightkube.models.meta_v1 import ObjectMeta
Quickstart
import asyncio
from lightkube import AsyncClient, ALL_NS
from lightkube.resources.core_v1 import Pod
async def list_all_pods():
client = AsyncClient() # Automatically loads kubeconfig or in-cluster config
try:
print("Listing all pods in all namespaces:")
async for pod in client.list(Pod, namespace=ALL_NS):
print(f" Pod: {pod.metadata.name}, Namespace: {pod.metadata.namespace}, Status: {pod.status.phase}")
except Exception as e:
print(f"An error occurred: {e}")
finally:
await client.close()
if __name__ == "__main__":
asyncio.run(list_all_pods())