kr8s Documentation

Responses from the kr8s Client API are returned as objects from the kr8s.objects (sync) or kr8s.asyncio.objects (async) modules. These objects represent Kubernetes resources (e.g., Pod, Service, Deployment).

• Sync usage: kr8s.get() returns a list of sync objects.
• Async usage: kr8s.asyncio.get() returns an async iterator of async objects.

Attributes and methods work identically in both sync and async versions, though async methods must be awaited.

import kr8s

# Sync
pods = list(kr8s.get("pods", namespace=kr8s.ALL))
pod = pods[0]

# Async
import kr8s.asyncio
pods = [po async for po in kr8s.asyncio.get("pods", namespace=kr8s.ALL)]
pod = pods[0]

When writing a kopf handler, the body argument contains the raw dictionary representation of the Kubernetes resource. To use kr8s features, you must wrap this body in a kr8s object class.

Synchronous approach: Use kr8s.objects.Pod.

Asynchronous approach: Use kr8s.asyncio.objects.Pod and await the instantiation.

# Sync
from kr8s.objects import Pod
pod = Pod(body)

# Async
from kr8s.asyncio.objects import Pod
pod = await Pod(body)

You can build event-driven Kubernetes controllers by combining kopf with kr8s. kopf handles the event lifecycle (creation, updates, deletions), while kr8s provides high-level Pythonic interaction with Kubernetes resources like Pods.

Core Workflow

1. Define Event Handlers: Use @kopf.on.create("resource_type") and @kopf.on.resume("resource_type") decorators to trigger logic.
2. Instantiate kr8s Objects: Convert the body provided by kopf into a kr8s object (e.g., Pod(body) for synchronous or await Pod(body) for asynchronous).
3. Perform Actions: Use kr8s methods like .exec(), .label(), or .exists() to interact with the cluster.
4. Run the Controller: Use python -m kopf run <your_file>.py to start the operator locally.

python -m kopf run controller.py

To interact with Kubernetes, kr8s uses an API Client. In most cases, you do not need to manage this object manually; calling high-level kr8s functions will automatically generate or retrieve a client from a cache.

Implicit Client Usage

You can call functions directly from the kr8s (sync) or kr8s.asyncio (async) modules without instantiating a client.

Explicit Client Usage

If you need to be explicit, you can construct an API client object using kr8s.api() or await kr8s.asyncio.api() and then call methods on that instance. You can also pass a specific client instance to resource objects (like Pod) during instantiation.

import kr8s

# Implicit (No client needed)
version = kr8s.version()

# Explicit (Sync)
api = kr8s.api()
version = api.version()

# Explicit (Async)
import kr8s.asyncio
api = await kr8s.asyncio.api()
version = await api.version()

To containerize your operator, use a Python base image and ensure kr8s is installed via pip. Copy your controller script into the image and set it as the CMD.

# Dockerfile
FROM python:3.11

WORKDIR /usr/local/src

RUN pip install kr8s

COPY controller.py /usr/local/src/

CMD ["python3", "/usr/local/src/controller.py"]

Build and push the image:

$ docker build -t foo/labelling-operator:latest .
$ docker push foo/labelling-operator:latest

To run a port forward without blocking your main execution flow, use the .start() and .stop() methods.

• In Sync code: .start() spawns the port forward in a background thread.
• In Async code: await .start() spawns the port forward as a background task. Warning: Your main code must be async and non-blocking because the port forward and your code share the same event loop.

from kr8s.objects import Pod

pod = Pod.get("my-pod")
pf = pod.portforward(remote_port=1234, local_port=5678)

# Starts the port forward in a background thread
pf.start()

# ... do other work ...

pf.stop()

If you want to use kr8s natively with asyncio or trio, use the kr8s.asyncio submodule. This provides direct access to asynchronous coroutines and async iterators. Many submodules have async equivalents, such as:

• kr8s.asyncio.objects (equivalent to kr8s.objects)
• kr8s.asyncio.portforward (equivalent to kr8s.portforward)

When using the async API, remember to use await for coroutines and async for for asynchronous iterators.

import kr8s.asyncio

pods = [pod async for pod in kr8s.asyncio.get("pods")]

Install the kr8s package from the conda-forge channel using conda.

$ conda install -c conda-forge kr8s