kr8s.asyncio.objects

Objects to represent Kubernetes resources.

This module provides classes that represent Kubernetes resources. These classes are used to interact with resources in the Kubernetes API server.

Classes

APIObject

Base class for Kubernetes objects.

Binding

A Kubernetes Binding.

ClusterRole

A Kubernetes ClusterRole.

ClusterRoleBinding

A Kubernetes ClusterRoleBinding.

ComponentStatus

A Kubernetes ComponentStatus.

ConfigMap

A Kubernetes ConfigMap.

ControllerRevision

A Kubernetes ControllerRevision.

CronJob

A Kubernetes CronJob.

CustomResourceDefinition

A Kubernetes CustomResourceDefinition.

DaemonSet

A Kubernetes DaemonSet.

Deployment

A Kubernetes Deployment.

Endpoints

A Kubernetes Endpoints.

Event

A Kubernetes Event.

HorizontalPodAutoscaler

A Kubernetes HorizontalPodAutoscaler.

Ingress

A Kubernetes Ingress.

IngressClass

A Kubernetes IngressClass.

Job

A Kubernetes Job.

LimitRange

A Kubernetes LimitRange.

Namespace

A Kubernetes Namespace.

NetworkPolicy

A Kubernetes NetworkPolicy.

Node

A Kubernetes Node.

PersistentVolume

A Kubernetes PersistentVolume.

PersistentVolumeClaim

A Kubernetes PersistentVolumeClaim.

Pod

A Kubernetes Pod.

PodDisruptionBudget

A Kubernetes PodDisruptionBudget.

PodTemplate

A Kubernetes PodTemplate.

ReplicaSet

A Kubernetes ReplicaSet.

ReplicationController

A Kubernetes ReplicationController.

ResourceQuota

A Kubernetes ResourceQuota.

Role

A Kubernetes Role.

RoleBinding

A Kubernetes RoleBinding.

Secret

A Kubernetes Secret.

Service

A Kubernetes Service.

ServiceAccount

A Kubernetes ServiceAccount.

StatefulSet

A Kubernetes StatefulSet.

Table

A Kubernetes Table.

Functions

get_class(→ type[APIObject])

Get an APIObject subclass by kind and version.

new_class(→ type[APIObject])

Create a new APIObject subclass.

object_from_name_type(→ APIObject)

Create an APIObject from a Kubernetes resource name.

object_from_spec(→ APIObject)

Create an APIObject from a Kubernetes resource spec.

objects_from_files(→ list[APIObject])

Create APIObjects from Kubernetes resource files.

Module Contents

class kr8s.asyncio.objects.APIObject(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Base class for Kubernetes objects.

version: str
endpoint: str
kind: str
plural: str
singular: str
namespaced: bool = False
scalable: bool = False
scalable_spec: str = 'replicas'
property api
property raw: Any

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.Binding(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes Binding.

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

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.ClusterRole(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes ClusterRole.

version = 'rbac.authorization.k8s.io/v1'
endpoint = 'clusterroles'
kind = 'ClusterRole'
plural = 'clusterroles'
singular = 'clusterrole'
namespaced = False
scalable: bool = False
scalable_spec: str = 'replicas'
property api
property raw: Any

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.ClusterRoleBinding(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes ClusterRoleBinding.

version = 'rbac.authorization.k8s.io/v1'
endpoint = 'clusterrolebindings'
kind = 'ClusterRoleBinding'
plural = 'clusterrolebindings'
singular = 'clusterrolebinding'
namespaced = False
scalable: bool = False
scalable_spec: str = 'replicas'
property api
property raw: Any

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.ComponentStatus(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes ComponentStatus.

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

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.ConfigMap(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes ConfigMap.

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

Data of the ConfigMap.

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

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.ControllerRevision(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes ControllerRevision.

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

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.CronJob(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes CronJob.

version = 'batch/v1'
endpoint = 'cronjobs'
kind = 'CronJob'
plural = 'cronjobs'
singular = 'cronjob'
namespaced = True
scalable: bool = False
scalable_spec: str = 'replicas'
property api
property raw: Any

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.CustomResourceDefinition(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes CustomResourceDefinition.

version = 'apiextensions.k8s.io/v1'
endpoint = 'customresourcedefinitions'
kind = 'CustomResourceDefinition'
plural = 'customresourcedefinitions'
singular = 'customresourcedefinition'
namespaced = False
scalable: bool = False
scalable_spec: str = 'replicas'
property api
property raw: Any

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.DaemonSet(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes DaemonSet.

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

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.Deployment(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes Deployment.

version = 'apps/v1'
endpoint = 'deployments'
kind = 'Deployment'
plural = 'deployments'
singular = 'deployment'
namespaced = True
scalable = True
async pods() list[Pod]

Return a list of Pods for this Deployment.

async ready()

Check if the deployment is ready.

scalable_spec: str = 'replicas'
property api
property raw: Any

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.Endpoints(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes Endpoints.

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

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.Event(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes Event.

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

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.HorizontalPodAutoscaler(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes HorizontalPodAutoscaler.

version = 'autoscaling/v2'
endpoint = 'horizontalpodautoscalers'
kind = 'HorizontalPodAutoscaler'
plural = 'horizontalpodautoscalers'
singular = 'horizontalpodautoscaler'
namespaced = True
scalable: bool = False
scalable_spec: str = 'replicas'
property api
property raw: Any

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.Ingress(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes Ingress.

version = 'networking.k8s.io/v1'
endpoint = 'ingresses'
kind = 'Ingress'
plural = 'ingresses'
singular = 'ingress'
namespaced = True
scalable: bool = False
scalable_spec: str = 'replicas'
property api
property raw: Any

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.IngressClass(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes IngressClass.

version = 'networking.k8s.io/v1'
endpoint = 'ingressclasses'
kind = 'IngressClass'
plural = 'ingressclasses'
singular = 'ingressclass'
namespaced = False
scalable: bool = False
scalable_spec: str = 'replicas'
property api
property raw: Any

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.Job(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes Job.

version = 'batch/v1'
endpoint = 'jobs'
kind = 'Job'
plural = 'jobs'
singular = 'job'
namespaced = True
scalable = True
scalable_spec = 'parallelism'
property api
property raw: Any

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.LimitRange(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes LimitRange.

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

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.Namespace(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes Namespace.

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

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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 = None, **kwargs) None

Annotate this object in Kubernetes.

async label(labels: dict | None = 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 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:

classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self]
Async:

List objects in Kubernetes.

Args:

api: An optional API object to use. **kwargs: Keyword arguments to pass to kr8s.get().

Returns:

A list of objects.

class kr8s.asyncio.objects.NetworkPolicy(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)

Bases: APIObject

A Kubernetes NetworkPolicy.

version = 'networking.k8s.io/v1'
endpoint = 'networkpolicies'
kind = 'NetworkPolicy'
plural = 'networkpolicies'
singular = 'networkpolicy'
namespaced = True
scalable: bool = False
scalable_spec: str = 'replicas'
property api
property raw: Any

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str | None

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 = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self
Async:

Get a Kubernetes resource by name or via selectors.

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, grace_period: int | None = None, force: bool = False) None

Delete this object from Kubernetes.

Args:

propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to True is equivelaent to setting grace_period to 0)

async refresh() None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int | None = None) None

Scale this object in Kubernetes.

async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]]

Watch this object in Kubernetes.

async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = 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
>>>