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#

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) aiohttp.ClientResponse#

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#