kr8s#

Submodules#

Package Contents#

Classes#

Api

A kr8s object for interacting with the Kubernetes API.

Functions#

get(*args, **kwargs)

Get a resource by name.

Attributes#

ALL

api

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) → 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 reauthenticate() → None#

Reauthenticate the API.

async get(kind: str, *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

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 watch(kind: str, namespace: str = None, label_selector: str | Dict = None, field_selector: str | Dict = None, since: str = None)#

Watch a Kubernetes resource.

async api_resources() → dict#

Get the Kubernetes API resources.

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#
kr8s.version#
kr8s.watch#
kr8s.api_resources#