kr8s

Submodules

Package Contents

Classes

Api

A kr8s object for interacting with the Kubernetes API.

Functions

get(*args, **kwargs)

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.

Attributes

ALL

version

watch

api_resources

kr8s.ALL = 'all'
exception kr8s.APITimeoutError

Bases: Exception

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

class args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception kr8s.ConnectionClosedError

Bases: Exception

A connection has been closed.

class args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception kr8s.ExecError

Bases: Exception

Internal error in the exec protocol.

class args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception kr8s.NotFoundError

Bases: Exception

Unable to find the requested resource.

class args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception kr8s.ServerError(message, status=None, response=None)

Bases: Exception

Error from the Kubernetes API server.

Attributes

statusstr

The Status object from the Kubernetes API server

responsehttpx.Response

The httpx response object

class args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

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.

property namespace: str

Get the default namespace.

async call_api(method: str = 'GET', version: str = 'v1', base: str = '', namespace: str = None, url: str = '', raise_for_status: bool = True, stream: bool = False, **kwargs) → httpx.Response

Make a Kubernetes API request.

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

Open a websocket connection to a Kubernetes API endpoint.

async version() → dict

Get the Kubernetes version information from the API.

Returns

dict

The Kubernetes version information.

async async_version() → dict
async reauthenticate() → None

Reauthenticate the API.

async whoami()

Retrieve the subject that’s currently authenticated.

Inspired by kubectl whoami.

Returns:

str: The subject that’s currently authenticated.

async async_whoami()
async async_get_kind(kind: str | type, namespace: str = None, label_selector: str | Dict = None, field_selector: str | Dict = None, params: dict = None, watch: bool = False, **kwargs) → dict

Get a Kubernetes resource.

async get(kind: str | type, *names: List[str], namespace: str = None, label_selector: str | Dict = None, field_selector: str | Dict = None, as_object: object = None, **kwargs) → List[object]

Get Kubernetes resources.

Parameters

kindstr, type

The kind of resource to get.

*namesList[str], optional

The names of specific resources to get.

namespacestr, optional

The namespace to get the resource from.

label_selectorUnion[str, Dict], optional

The label selector to filter the resources by.

field_selectorUnion[str, Dict], optional

The field selector to filter the resources by.

as_objectobject, optional

The object to return the resources as.

**kwargs

Additional keyword arguments to pass to the API call.

Returns

List[object]

The resources.

async async_get(kind: str | type, *names: List[str], namespace: str = None, label_selector: str | Dict = None, field_selector: str | Dict = None, as_object: object = None, **kwargs) → List[object]
async watch(kind: str, namespace: str = None, label_selector: str | Dict = None, field_selector: str | Dict = None, since: str = None)

Watch a Kubernetes resource.

async async_watch(kind: str, namespace: str = None, label_selector: str | Dict = None, field_selector: str | Dict = None, since: str = None) → Tuple[str, object]

Watch a Kubernetes resource.

async api_resources() → dict

Get the Kubernetes API resources.

async async_api_resources() → dict

Get the Kubernetes API resources.

async api_versions() → List[str]

Get the Kubernetes API versions.

async async_api_versions() → List[str]
kr8s.get(*args, **kwargs)

Get a resource by name.

Parameters

kindstr

The kind of resource to get

*namesList[str]

The names of the resources to get

namespacestr, optional

The namespace to get the resource from

label_selectorUnion[str, Dict], optional

The label selector to filter the resources by

field_selectorUnion[str, Dict], optional

The field selector to filter the resources by

as_objectobject, optional

The object to populate with the resource data

apiApi, optional

The api to use to get the resource

Returns

object

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, kubeconfig: str = None, serviceaccount: str = None, namespace: str = None, context: str = 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.

Parameters

urlstr, optional

The URL of the Kubernetes API server

kubeconfigstr, optional

The path to a kubeconfig file to use

serviceaccountstr, optional

The path of a service account to use

namespacestr, optional

The namespace to use

contextstr, optional

The context to use

Returns

Api

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

str

The user’s identity

Examples

>>> import kr8s
>>> print(kr8s.whoami())
kr8s.version
kr8s.watch
kr8s.api_resources