kr8s.objects

Attributes

Classes

APIObject

Base class for Kubernetes objects.

Binding

A Kubernetes Binding.

ComponentStatus

A Kubernetes ComponentStatus.

ConfigMap

A Kubernetes ConfigMap.

Endpoints

A Kubernetes Endpoints.

Event

A Kubernetes Event.

LimitRange

A Kubernetes LimitRange.

Namespace

A Kubernetes Namespace.

Node

A Kubernetes Node.

PersistentVolume

A Kubernetes PersistentVolume.

PersistentVolumeClaim

A Kubernetes PersistentVolumeClaim.

Pod

A Kubernetes Pod.

PodTemplate

A Kubernetes PodTemplate.

ReplicationController

A Kubernetes ReplicationController.

ResourceQuota

A Kubernetes ResourceQuota.

Secret

A Kubernetes Secret.

Service

A Kubernetes Service.

ServiceAccount

A Kubernetes ServiceAccount.

ControllerRevision

A Kubernetes ControllerRevision.

DaemonSet

A Kubernetes DaemonSet.

Deployment

A Kubernetes Deployment.

ReplicaSet

A Kubernetes ReplicaSet.

StatefulSet

A Kubernetes StatefulSet.

HorizontalPodAutoscaler

A Kubernetes HorizontalPodAutoscaler.

CronJob

A Kubernetes CronJob.

Job

A Kubernetes Job.

Ingress

A Kubernetes Ingress.

IngressClass

A Kubernetes IngressClass.

NetworkPolicy

A Kubernetes NetworkPolicy.

PodDisruptionBudget

A Kubernetes PodDisruptionBudget.

ClusterRoleBinding

A Kubernetes ClusterRoleBinding.

ClusterRole

A Kubernetes ClusterRole.

RoleBinding

A Kubernetes RoleBinding.

Role

A Kubernetes Role.

CustomResourceDefinition

A Kubernetes CustomResourceDefinition.

Table

A Kubernetes Table.

Functions

run_sync(→ Callable[P, T])

Wraps a coroutine in a function that blocks until it has executed.

sync(→ C)

Convert all public async methods/properties of an object to universal methods.

Module Contents

kr8s.objects.run_sync(coro: Callable[P, Awaitable[T]]) Callable[P, T]

Wraps a coroutine in a function that blocks until it has executed.

Args:

coro (Awaitable): A coroutine.

Returns:

Callable: A sync function that executes the coroutine via the :class`Portal`.

kr8s.objects.sync(source: C) C

Convert all public async methods/properties of an object to universal methods.

Private methods or methods starting with “async_” are ignored. See run_sync() for more info on how the conversion works.

Args:

source (C): object with coroutines to convert

Returns:

C: converted object with sync methods

Examples:

It’s common to implement a coroutine and name it with async_ and then wrap that in another corotine that calls it. That way the method can be called from both sync and async contexts as function allows you to convert the outer coroutine to a sync method. But other async methods or other async objects can call the inner coroutine directly.

>>> class Foo:
...     async def async_bar(self):
...         return 42
...     async def bar(self):
...         return await self.async_bar()
...     async def baz(self):
...         # If you want to calll self.bar() from another async method
...         # you can't when it gets wrapped with sync, so you can call
...         # self.async_bar() directly instead.
...         return (await self.async_bar()) + 1
...
>>> SyncFoo = sync(Foo)
>>> inst = SyncFoo()
>>> inst.bar()
42
>>> inst.async_bar()
<coroutine object Foo.async_bar at 0x7fbe0442b940>
>>> inst.baz()
43
class kr8s.objects.APIObject(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.APIObject

Base class for Kubernetes objects.

namespaced = False
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.Binding(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.Binding

A Kubernetes Binding.

version = 'v1'
endpoint = 'bindings'
kind = 'Binding'
plural = 'bindings'
singular = 'binding'
namespaced = True
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.ComponentStatus(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.ComponentStatus

A Kubernetes ComponentStatus.

version = 'v1'
endpoint = 'componentstatuses'
kind = 'ComponentStatus'
plural = 'componentstatuses'
singular = 'componentstatus'
namespaced = False
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.ConfigMap(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.ConfigMap

A Kubernetes ConfigMap.

version = 'v1'
endpoint = 'configmaps'
kind = 'ConfigMap'
plural = 'configmaps'
singular = 'configmap'
namespaced = True
property data: box.Box

Data of the ConfigMap.

scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.Endpoints(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.Endpoints

A Kubernetes Endpoints.

version = 'v1'
endpoint = 'endpoints'
kind = 'Endpoints'
plural = 'endpoints'
singular = 'endpoint'
namespaced = True
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.Event(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.Event

A Kubernetes Event.

version = 'v1'
endpoint = 'events'
kind = 'Event'
plural = 'events'
singular = 'event'
namespaced = True
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.LimitRange(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.LimitRange

A Kubernetes LimitRange.

version = 'v1'
endpoint = 'limitranges'
kind = 'LimitRange'
plural = 'limitranges'
singular = 'limitrange'
namespaced = True
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.Namespace(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.Namespace

A Kubernetes Namespace.

version = 'v1'
endpoint = 'namespaces'
kind = 'Namespace'
plural = 'namespaces'
singular = 'namespace'
namespaced = False
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.Node(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.Node

A Kubernetes Node.

version = 'v1'
endpoint = 'nodes'
kind = 'Node'
plural = 'nodes'
singular = 'node'
namespaced = False
property unschedulable
async cordon() None

Cordon the node.

This will mark the node as unschedulable.

async uncordon() None

Uncordon the node.

This will mark the node as schedulable.

scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.PersistentVolume(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.PersistentVolume

A Kubernetes PersistentVolume.

version = 'v1'
endpoint = 'persistentvolumes'
kind = 'PersistentVolume'
plural = 'persistentvolumes'
singular = 'persistentvolume'
namespaced = False
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.PersistentVolumeClaim(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.PersistentVolumeClaim

A Kubernetes PersistentVolumeClaim.

version = 'v1'
endpoint = 'persistentvolumeclaims'
kind = 'PersistentVolumeClaim'
plural = 'persistentvolumeclaims'
singular = 'persistentvolumeclaim'
namespaced = True
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.Pod(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.Pod

A Kubernetes Pod.

version = 'v1'
endpoint = 'pods'
kind = 'Pod'
plural = 'pods'
singular = 'pod'
namespaced = True
async async_ready() bool

Check if the pod is ready.

async ready() bool

Check if the pod is ready.

async logs(container=None, pretty=None, previous=False, since_seconds=None, since_time=None, timestamps=False, tail_lines=None, limit_bytes=None, follow=False, timeout=3600) AsyncGenerator[str, None, None]

Streams logs from a Pod.

Args:
container:

The container to get logs from. Defaults to the first container in the Pod.

pretty:

If True, return pretty logs. Defaults to False.

previous:

If True, return previous terminated container logs. Defaults to False.

since_seconds:

If set, return logs since this many seconds ago.

since_time:

If set, return logs since this time.

timestamps:

If True, prepend each log line with a timestamp. Defaults to False.

tail_lines:

If set, return this many lines from the end of the logs.

limit_bytes:

If set, return this many bytes from the end of the logs.

follow:

If True, follow the logs until the timeout is reached. Defaults to False.

timeout:

If following timeout after this many seconds. Set to None to disable timeout.

Returns:

An async generator yielding log lines.

Example:
>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> for line in pod.logs():
...     print(line)

We can also follow logs as they are generated, the generator will yield a new log line as it is generated by the Pod. This blocks indefinitely so we can set a timeout to break after some period of time, the default is 3600 (1hr) but can be set to None to disable the timeout.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod", namespace="ns")
>>> for line in pod.logs(follow=True, timeout=60):
...     # Will continue streaming logs until 60 seconds or the Pod is terminated
...     print(line)
portforward(remote_port: int, local_port: int = None, address: List[str] | str = '127.0.0.1') int

Port forward a pod.

Returns an instance of kr8s.portforward.PortForward for this Pod.

Example:

This can be used as a an async context manager or with explicit start/stop methods.

Context manager:

>>> async with pod.portforward(8888) as port:
...     print(f"Forwarding to port {port}")
...     # Do something with port 8888

Explict start/stop:

>>> pf = pod.portforward(8888)
>>> await pf.start()
>>> print(f"Forwarding to port {pf.local_port}")
>>> # Do something with port 8888
>>> await pf.stop()

Explict bind address:

>>> async with pod.PortForward(8888, address=["127.0.0.1", "10.20.0.1"]) as port:
...     print(f"Forwarding to port {port}")
...     # Do something with port 8888 on the Pod, port will be bind to 127.0.0.1 and 10.20.0.1
async async_exec(command: List[str], *, container: str = None, stdin: Union(str | bytes | BinaryIO) = None, stdout: BinaryIO = None, stderr: BinaryIO = None, check: bool = True, capture_output: bool = True)
async exec(command: List[str], *, container: str = None, stdin: Union(str | bytes | BinaryIO) = None, stdout: BinaryIO = None, stderr: BinaryIO = None, check: bool = True, capture_output: bool = True)

Run a command in a container and wait until it completes.

Behaves like subprocess.run().

Args:
command:

Command to execute.

container:

Container to execute the command in.

stdin:

If set, read stdin to the container.

stdout:

If set, write stdout to the provided writable stream object.

stderr:

If set, write stderr to the provided writable stream object.

check:

If True, raise an exception if the command fails.

capture_output:

If True, store stdout and stderr from the container in an attribute.

Returns:

A kr8s._exec.CompletedExec object.

Example:
>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> ex = await pod.exec(["ls", "-l"])
>>> print(ex.stdout)
>>> print(ex.stderr)
classmethod gen(*, name, image, namespace=None, annotations=None, command=None, env=None, resources=None, image_pull_policy=None, labels=None, ports=None, restart='Always')

Generate a pod definition.

Args:

name (str): The name of the pod. namespace (str): The namespace of the pod. image (str): The image to use. annotations (dict): Annotations to add to the pod. command (list): Command to run in the container. env (dict): Environment variables to set in the container. resources (dict): Resources to set in the container. image_pull_policy (str): Image pull policy to use. labels (dict): Labels to add to the pod. ports (list|int): Ports to expose. restart (str): Restart policy to use.

Returns:

A kr8s.objects.Pod object.

Example:
>>> from kr8s.objects import Pod
>>> pod = Pod.gen(name="my-pod", image="my-image")
>>> pod.create()

Create an nginx Pod that exposes port 80. >>> from kr8s.objects import Pod >>> pod = Pod.gen(name=”nginx”, image=”nginx:latest”, ports=[80]) >>> pod.create()

Create an wordpress Pod that exposes port 80. >>> from kr8s.objects import Pod >>> pod = Pod.gen(name=”wordpress”, image=”wordpress:latest”, ports=[{“containerPort”: 80}]) >>> pod.create()

Create a Pod that requires a GPU >>> from kr8s.objects import Pod >>> pod = Pod.gen(name=”cuda-vectoradd”, … image=”nvidia/samples:vectoradd-cuda11.6.0-ubuntu18.04”, … resources={“limits”: {“nvidia.com/gpu”: 1}})

scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
class kr8s.objects.PodTemplate(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.PodTemplate

A Kubernetes PodTemplate.

version = 'v1'
endpoint = 'podtemplates'
kind = 'PodTemplate'
plural = 'podtemplates'
singular = 'podtemplate'
namespaced = True
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.ReplicationController(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.ReplicationController

A Kubernetes ReplicationController.

version = 'v1'
endpoint = 'replicationcontrollers'
kind = 'ReplicationController'
plural = 'replicationcontrollers'
singular = 'replicationcontroller'
namespaced = True
scalable = True
async ready()

Check if the deployment is ready.

scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.ResourceQuota(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.ResourceQuota

A Kubernetes ResourceQuota.

version = 'v1'
endpoint = 'resourcequotas'
kind = 'ResourceQuota'
plural = 'resourcequotas'
singular = 'resourcequota'
namespaced = True
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.Secret(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.Secret

A Kubernetes Secret.

version = 'v1'
endpoint = 'secrets'
kind = 'Secret'
plural = 'secrets'
singular = 'secret'
namespaced = True
property data: box.Box

Data of the Secret.

scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.Service(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.Service

A Kubernetes Service.

version = 'v1'
endpoint = 'services'
kind = 'Service'
plural = 'services'
singular = 'service'
namespaced = True
async proxy_http_request(method: str, path: str, port: int | None = None, **kwargs: Any) httpx.Response

Issue a HTTP request with specific HTTP method to proxy of a Service.

Args:

method: HTTP method to use. path: Path to proxy. port: Port to proxy to. If not specified, the first port in the

Service’s spec will be used.

**kwargs: Additional keyword arguments to pass to the API call.

async async_proxy_http_request(method: str, path: str, port: int | None = None, **kwargs: Any) httpx.Response
async proxy_http_get(path: str, port: int | None = None, **kwargs) httpx.Response
async proxy_http_post(path: str, port: int | None = None, **kwargs) None
async proxy_http_put(path: str, port: int | None = None, **kwargs) httpx.Response
async proxy_http_delete(path: str, port: int | None = None, **kwargs) httpx.Response
async ready_pods() List[Pod]

Return a list of ready Pods for this Service.

async async_ready_pods() List[Pod]

Return a list of ready Pods for this Service.

async ready() bool

Check if the service is ready.

portforward(remote_port: int, local_port: int = None, address: str | List[str] = '127.0.0.1') int

Port forward a service.

Returns an instance of kr8s.portforward.PortForward for this Service.

Example:

This can be used as a an async context manager or with explicit start/stop methods.

Context manager:

>>> async with service.portforward(8888) as port:
...     print(f"Forwarding to port {port}")
...     # Do something with port 8888

Explict start/stop:

>>> pf = service.portforward(8888)
>>> await pf.start()
>>> print(f"Forwarding to port {pf.local_port}")
>>> # Do something with port 8888
>>> await pf.stop()
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.ServiceAccount(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.ServiceAccount

A Kubernetes ServiceAccount.

version = 'v1'
endpoint = 'serviceaccounts'
kind = 'ServiceAccount'
plural = 'serviceaccounts'
singular = 'serviceaccount'
namespaced = True
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.ControllerRevision(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.ControllerRevision

A Kubernetes ControllerRevision.

version = 'apps/v1'
endpoint = 'controllerrevisions'
kind = 'ControllerRevision'
plural = 'controllerrevisions'
singular = 'controllerrevision'
namespaced = True
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app")
keys() list

Return the keys of this object.

async set_owner(owner: APIObject) None

Set the owner reference of this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

owner: The owner object to set a reference to.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)
async async_set_owner(owner: APIObject) None

Set the owner of this object.

async adopt(child: APIObject) None

Adopt this object.

This will set the owner reference of the child object to this object.

See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/

Args:

child: The child object to adopt.

Example:
>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> deployment.adopt(pod)
to_dict() dict

Return a dictionary representation of this object.

to_lightkube() Any

Return a lightkube representation of this object.

to_pykube(api) Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)
classmethod gen(*args, **kwargs)
Abstractmethod:

class kr8s.objects.DaemonSet(resource: dict, namespace: str = None, api: kr8s._api.Api = None)

Bases: kr8s._objects.DaemonSet

A Kubernetes DaemonSet.

version = 'apps/v1'
endpoint = 'daemonsets'
kind = 'DaemonSet'
plural = 'daemonsets'
singular = 'daemonset'
namespaced = True
scalable = False
scalable_spec = 'replicas'
property api
property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

classmethod get(name: str = None, namespace: str = None, api: kr8s._api.Api = None, label_selector: str | Dict[str, str] = None, field_selector: str | Dict[str, str] = None, timeout: int = 2, **kwargs) APIObjectType
Async:

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) bool

Check if this object exists in Kubernetes.

async create() None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) None

Delete this object from Kubernetes.

async refresh() None

Refresh this object from Kubernetes.

async async_refresh() None

Refresh this object from Kubernetes.

async patch(patch, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async async_patch(patch: Dict, *, subresource=None, type=None) None

Patch this object in Kubernetes.

async scale(replicas: int = None) None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

classmethod list(**kwargs) List[APIObjectType]
Async:

List objects in Kubernetes.

Args:

**kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

async wait(conditions: List[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int = None)

Wait for conditions to be met.

Args:

conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.

Example:

Wait for a Pod to be ready.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait("condition=Ready")

Wait for a Job to either succeed or fail.

>>> from kr8s.objects import Job
>>> job = Job.get("my-jod")
>>> job.wait(["condition=Complete", "condition=Failed"])

Wait for a Pod to be initialized and ready to start containers.

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
Note:

As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.

Given that for is a reserved word anyway we can’t exactly match the kubectl API so we use condition in combination with a mode instead.

async annotate(annotations: dict = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict = None, **kwargs) None

Add labels to this object in Kubernetes.

Labels can be passed as a dictionary or as keyword arguments.

Args:
labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:
>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Both of these are equivalent
>>> deployment.label({"app": "my-app"})
>>> deployment.label(app="my-app"