kr8s.asyncio

The kr8s asynchronous API.

This module provides an asynchronous API for interacting with a Kubernetes cluster.

Submodules

Classes

Api

A kr8s object for interacting with the Kubernetes API.

Functions

api(→ kr8s._api.Api)

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

api_resources([api, _asyncio])

get(kind, *names[, namespace, label_selector, ...])

Get a resource by name.

version([api, _asyncio])

watch(kind[, namespace, label_selector, ...])

whoami([api, _asyncio])

Package Contents

class kr8s.asyncio.Api(**kwargs)

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.

auth
property timeout
async 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) AsyncGenerator[httpx.Response]

Make a Kubernetes API request.

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

Open a websocket connection to a Kubernetes API endpoint.

async version() dict

Get the Kubernetes version information from the API.

Returns:

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_lookup_kind(kind) tuple[str, str, bool]

Lookup a Kubernetes resource kind.

async 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.

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) AsyncGenerator[tuple[type[kr8s._objects.APIObject], httpx.Response]]

Get a Kubernetes resource.

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) kr8s._objects.APIObject | list[kr8s._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.

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) kr8s._objects.APIObject | list[kr8s._objects.APIObject]
async watch(kind: str, namespace: str | None = None, label_selector: str | dict | None = None, field_selector: str | dict | None = None, since: str | None = None)

Watch a Kubernetes resource.

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) AsyncGenerator[tuple[str, kr8s._objects.APIObject]]

Watch a Kubernetes resource.

async api_resources() list[dict]

Get the Kubernetes API resources.

async async_api_resources() list[dict]

Get the Kubernetes API resources.

async api_versions() AsyncGenerator[str]

Get the Kubernetes API versions.

async async_api_versions() AsyncGenerator[str]
property namespace: str

Get the default namespace.

async kr8s.asyncio.api(url: str | None = None, kubeconfig: str | None = None, serviceaccount: str | None = None, namespace: str | None = None, context: str | None = None, _asyncio: bool = True) kr8s._api.Api

Create a kr8s.asyncio.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:

kr8s.asyncio.Api: The API object

Examples:
>>> import kr8s
>>> api = await kr8s.asyncio.api()  # Uses the default kubeconfig
>>> print(await api.version())  # Get the Kubernetes version
async kr8s.asyncio.api_resources(api=None, _asyncio=True)
async kr8s.asyncio.get(kind: str, *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, api=None, _asyncio=True, **kwargs)

Get a resource by name.

This function retrieves a resource from the Kubernetes cluster based on its kind and name(s). It supports various options for filtering and customization.

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: Automatically create a class for the resource if none exists. api: The api to use to get the resource. **kwargs: Additional keyword arguments to pass to the httpx API call.

Returns:

List[APIObject]: The Kubernetes resource objects.

Raises:

ValueError: If the resource is not found.

Examples:
>>> import kr8s
>>> # All of these are equivalent
>>> ings = await kr8s.asyncio.get("ing")                           # Short name
>>> ings = await kr8s.asyncio.get("ingress")                       # Singular
>>> ings = await kr8s.asyncio.get("ingresses")                     # Plural
>>> ings = await kr8s.asyncio.get("Ingress")                       # Title
>>> ings = await kr8s.asyncio.get("ingress.networking.k8s.io")     # Full group name
>>> ings = await kr8s.asyncio.get("ingress.v1.networking.k8s.io")  # Full with explicit version
>>> ings = await kr8s.asyncio.get("ingress.networking.k8s.io/v1")  # Full with explicit version alt.
async kr8s.asyncio.version(api=None, _asyncio=True)
async kr8s.asyncio.watch(kind: str, namespace: str | None = None, label_selector: str | Dict | None = None, field_selector: str | Dict | None = None, since: str | None = None, api=None, _asyncio=True)
async kr8s.asyncio.whoami(api=None, _asyncio=True)