kr8s

This module contains kr8s, a simple, extensible Python client library for Kubernetes.

At the top level, kr8s provides a synchronous API that wraps the asynchronous API provided by kr8s.asyncio. Both APIs are functionally identical with the same objects, method signatures and return values.

Submodules

• kr8s.asyncio
• kr8s.objects
• kr8s.portforward

Attributes

ALL
version
watch
api_resources

Exceptions

APITimeoutError	A timeout has occurred while waiting for a response from the Kubernetes API server.
ConnectionClosedError	A connection has been closed.
ExecError	Internal error in the exec protocol.
NotFoundError	Unable to find the requested resource.
ServerError	Error from the Kubernetes API server.

Classes

Api	A kr8s object for interacting with the Kubernetes API.

Functions

get(kind, *names[, namespace, label_selector, ...])	Get a resource by name.
api(→ Api)	Create a kr8s.Api object for interacting with the Kubernetes API.
whoami()	Get the current user's identity.

Package Contents

kr8s.ALL = 'all'

exception kr8s.APITimeoutError

Bases: Exception

A timeout has occurred while waiting for a response from the Kubernetes API server.

exception kr8s.ConnectionClosedError

Bases: Exception

A connection has been closed.

exception kr8s.ExecError

Bases: Exception

Internal error in the exec protocol.

exception kr8s.NotFoundError

Bases: Exception

Unable to find the requested resource.

exception kr8s.ServerError(message: str, status: str | None = None, response: httpx.Response | None = None)

Bases: Exception

Error from the Kubernetes API server.

Attributes:

status: The Status object from the Kubernetes API server response: The httpx response object

status = None

response = None

class kr8s.Api(**kwargs)

Bases: _api.Api

A kr8s object for interacting with the Kubernetes API.

Warning

This class is not intended to be instantiated directly. Instead, use the kr8s.api() function to get a singleton instance of the API.

See https://docs.kr8s.org/en/stable/client.html#client-caching.

version() → dict

Get the Kubernetes version information from the API.

Returns:

The Kubernetes version information.

reauthenticate()

Reauthenticate the API.

whoami()

Retrieve the subject that’s currently authenticated.

Inspired by kubectl whoami.

Returns:

str: The subject that’s currently authenticated.

lookup_kind(kind) → tuple[str, str, bool]

Lookup a Kubernetes resource kind.

Check whether a resource kind exists on the remote server.

Args:

kind: The kind of resource to lookup.

Returns:

The kind of resource, the plural form and whether the resource is namespaced

Raises:

ValueError: If the kind is not found.

get(kind: str | type, *names: str, namespace: str | None = None, label_selector: str | dict | None = None, field_selector: str | dict | None = None, as_object: type[_objects.APIObject] | None = None, allow_unknown_type: bool = True, **kwargs) → collections.abc.Generator[objects.APIObject]

Get Kubernetes resources.

Args:

kind: The kind of resource to get. *names: The names of specific resources to get. namespace: The namespace to get the resource from. label_selector: The label selector to filter the resources by. field_selector: The field selector to filter the resources by. as_object: The object to return the resources as. allow_unknown_type: Automatically create a class for the resource if none exists, default True. **kwargs: Additional keyword arguments to pass to the API call.

Returns:

The resources.

watch(kind: str, namespace: str | None = None, label_selector: str | dict | None = None, field_selector: str | dict | None = None, since: str | None = None) → collections.abc.Generator[tuple[str, objects.APIObject]]

Watch a Kubernetes resource.

api_resources() → list[dict]

Get the Kubernetes API resources.

api_versions() → collections.abc.Generator[str]

Get the Kubernetes API versions.

auth

property timeout

call_api(method: str = 'GET', version: str = 'v1', base: str = '', namespace: str | None = None, url: str = '', raise_for_status: bool = True, stream: bool = False, **kwargs) → collections.abc.AsyncGenerator[httpx.Response]

Make a Kubernetes API request.

open_websocket(version: str = 'v1', base: str = '', namespace: str | None = None, url: str = '', **kwargs) → collections.abc.AsyncGenerator[httpx_ws.AsyncWebSocketSession]

Open a websocket connection to a Kubernetes API endpoint.

async async_version() → dict

async async_reauthenticate() → None

async async_whoami()

async async_lookup_kind(kind) → tuple[str, str, bool]

Lookup a Kubernetes resource kind.

async async_get_kind(kind: str | type[kr8s._objects.APIObject], namespace: str | None = None, label_selector: str | dict | None = None, field_selector: str | dict | None = None, params: dict | None = None, watch: bool = False, allow_unknown_type: bool = True, **kwargs) → collections.abc.AsyncGenerator[tuple[type[kr8s._objects.APIObject], httpx.Response]]

Get a Kubernetes resource.

async async_get(kind: str | type, *names: str, namespace: str | None = None, label_selector: str | dict | None = None, field_selector: str | dict | None = None, as_object: type[kr8s._objects.APIObject] | None = None, allow_unknown_type: bool = True, **kwargs) → collections.abc.AsyncGenerator[kr8s._objects.APIObject]

async async_watch(kind: str, namespace: str | None = None, label_selector: str | dict | None = None, field_selector: str | dict | None = None, since: str | None = None, allow_unknown_type: bool = True) → collections.abc.AsyncGenerator[tuple[str, kr8s._objects.APIObject]]

Watch a Kubernetes resource.

async async_api_resources() → list[dict]

Get the Kubernetes API resources.

async async_api_versions() → collections.abc.AsyncGenerator[str]

property namespace: str

Get the default namespace.

kr8s.get(kind: str, *names: str, namespace: str | None = None, label_selector: str | dict | None = None, field_selector: str | dict | None = None, as_object: type | None = None, allow_unknown_type: bool = True, api=None, **kwargs)

Get a resource by name.

Args:

kind: The kind of resource to get *names: The names of the resources to get namespace: The namespace to get the resource from label_selector: The label selector to filter the resources by field_selector: The field selector to filter the resources by as_object: The object to populate with the resource data allow_unknown_type: Whether to allow unknown types api: The api to use to get the resource **kwargs: Additional arguments to pass to the API

Returns:

The populated object

Raises:

ValueError: If the resource is not found

Examples:

>>> import kr8s
>>> # All of these are equivalent
>>> ings = kr8s.get("ing")                           # Short name
>>> ings = kr8s.get("ingress")                       # Singular
>>> ings = kr8s.get("ingresses")                     # Plural
>>> ings = kr8s.get("Ingress")                       # Title
>>> ings = kr8s.get("ingress.networking.k8s.io")     # Full group name
>>> ings = kr8s.get("ingress.v1.networking.k8s.io")  # Full with explicit version
>>> ings = kr8s.get("ingress.networking.k8s.io/v1")  # Full with explicit version alt.

kr8s.api(url: str | None = None, kubeconfig: str | None = None, serviceaccount: str | None = None, namespace: str | None = None, context: str | None = None) → Api

Create a kr8s.Api object for interacting with the Kubernetes API.

If a kr8s object already exists with the same arguments in this thread, it will be returned.

Args:

url: The URL of the Kubernetes API server kubeconfig: The path to a kubeconfig file to use serviceaccount: The path of a service account to use namespace: The namespace to use context: The context to use

Returns:

The API object

Examples:

>>> import kr8s
>>> api = kr8s.api()  # Uses the default kubeconfig
>>> print(api.version())  # Get the Kubernetes version

kr8s.whoami()

Get the current user’s identity.

Returns:

The user’s identity

Examples:

>>> import kr8s
>>> print(kr8s.whoami())

kr8s.version

kr8s.watch

kr8s.api_resources