kr8s.objects

Module Contents

Classes

APIObject	Base class for Kubernetes objects.
Binding	A Kubernetes Binding.
ComponentStatus	A Kubernetes ComponentStatus.
ConfigMap	A Kubernetes ConfigMap.
Endpoints	A Kubernetes Endpoints.
Event	A Kubernetes Event.
LimitRange	A Kubernetes LimitRange.
Namespace	A Kubernetes Namespace.
Node	A Kubernetes Node.
PersistentVolume	A Kubernetes PersistentVolume.
PersistentVolumeClaim	A Kubernetes PersistentVolumeClaim.
Pod	A Kubernetes Pod.
PodTemplate	A Kubernetes PodTemplate.
ReplicationController	A Kubernetes ReplicationController.
ResourceQuota	A Kubernetes ResourceQuota.
Secret	A Kubernetes Secret.
Service	A Kubernetes Service.
ServiceAccount	A Kubernetes ServiceAccount.
ControllerRevision	A Kubernetes ControllerRevision.
DaemonSet	A Kubernetes DaemonSet.
Deployment	A Kubernetes Deployment.
ReplicaSet	A Kubernetes ReplicaSet.
StatefulSet	A Kubernetes StatefulSet.
HorizontalPodAutoscaler	A Kubernetes HorizontalPodAutoscaler.
CronJob	A Kubernetes CronJob.
Job	A Kubernetes Job.
Ingress	A Kubernetes Ingress.
IngressClass	A Kubernetes IngressClass.
NetworkPolicy	A Kubernetes NetworkPolicy.
PodDisruptionBudget	A Kubernetes PodDisruptionBudget.
ClusterRoleBinding	A Kubernetes ClusterRoleBinding.
ClusterRole	A Kubernetes ClusterRole.
RoleBinding	A Kubernetes RoleBinding.
Role	A Kubernetes Role.
CustomResourceDefinition	A Kubernetes CustomResourceDefinition.
Table	A Kubernetes Table.

Attributes

object_from_name_type
objects_from_files

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

Bases: kr8s._objects.APIObject

Base class for Kubernetes objects.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

namespaced = False

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.Binding

A Kubernetes Binding.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'bindings'

kind = 'Binding'

plural = 'bindings'

singular = 'binding'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.ComponentStatus

A Kubernetes ComponentStatus.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'componentstatuses'

kind = 'ComponentStatus'

plural = 'componentstatuses'

singular = 'componentstatus'

namespaced = False

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.ConfigMap

A Kubernetes ConfigMap.

property data: box.Box

Data of the ConfigMap.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'configmaps'

kind = 'ConfigMap'

plural = 'configmaps'

singular = 'configmap'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.Endpoints

A Kubernetes Endpoints.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'endpoints'

kind = 'Endpoints'

plural = 'endpoints'

singular = 'endpoint'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.Event

A Kubernetes Event.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'events'

kind = 'Event'

plural = 'events'

singular = 'event'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.LimitRange

A Kubernetes LimitRange.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'limitranges'

kind = 'LimitRange'

plural = 'limitranges'

singular = 'limitrange'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.Namespace

A Kubernetes Namespace.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'namespaces'

kind = 'Namespace'

plural = 'namespaces'

singular = 'namespace'

namespaced = False

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.Node

A Kubernetes Node.

property unschedulable

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'nodes'

kind = 'Node'

plural = 'nodes'

singular = 'node'

namespaced = False

scalable = False

scalable_spec = 'replicas'

async cordon() → None

Cordon the node.

This will mark the node as unschedulable.

async uncordon() → None

Uncordon the node.

This will mark the node as schedulable.

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.PersistentVolume

A Kubernetes PersistentVolume.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'persistentvolumes'

kind = 'PersistentVolume'

plural = 'persistentvolumes'

singular = 'persistentvolume'

namespaced = False

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.PersistentVolumeClaim

A Kubernetes PersistentVolumeClaim.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'persistentvolumeclaims'

kind = 'PersistentVolumeClaim'

plural = 'persistentvolumeclaims'

singular = 'persistentvolumeclaim'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.Pod

A Kubernetes Pod.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'pods'

kind = 'Pod'

plural = 'pods'

singular = 'pod'

namespaced = True

scalable = False

scalable_spec = 'replicas'

async async_ready() → bool

Check if the pod is ready.

async ready() → bool

Check if the pod is ready.

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

Streams logs from a Pod.

Args:

container:

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

pretty:

If True, return pretty logs. Defaults to False.

previous:

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

since_seconds:

If set, return logs since this many seconds ago.

since_time:

If set, return logs since this time.

timestamps:

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

tail_lines:

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

limit_bytes:

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

follow:

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

timeout:

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

Returns:

An async generator yielding log lines.

Example:

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

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

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod", namespace="ns")
>>> for line in pod.logs(follow=True, timeout=60):
...     # Will continue streaming logs until 60 seconds or the Pod is terminated
...     print(line)

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

Port forward a pod.

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

Example:

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

Context manager:

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

Explict start/stop:

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

Explict bind address:

>>> async with pod.PortForward(8888, address=["127.0.0.1", "10.20.0.1"]) as port:
...     print(f"Forwarding to port {port}")
...     # Do something with port 8888 on the Pod, port will be bind to 127.0.0.1 and 10.20.0.1

async async_exec(command: List[str], *, container: str = None, stdin: Union(str | bytes | BinaryIO) = None, stdout: BinaryIO = None, stderr: BinaryIO = None, check: bool = True, capture_output: bool = True)

async exec(command: List[str], *, container: str = None, stdin: Union(str | bytes | BinaryIO) = None, stdout: BinaryIO = None, stderr: BinaryIO = None, check: bool = True, capture_output: bool = True)

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

Behaves like subprocess.run().

Args:

command:

Command to execute.

container:

Container to execute the command in.

stdin:

If set, read stdin to the container.

stdout:

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

stderr:

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

check:

If True, raise an exception if the command fails.

capture_output:

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

Returns:

A kr8s._exec.CompletedExec object.

Example:

>>> from kr8s.objects import Pod
>>> pod = Pod.get("my-pod")
>>> ex = await pod.exec(["ls", "-l"])
>>> print(ex.stdout)
>>> print(ex.stderr)

classmethod gen(*, name, image, namespace=None, annotations=None, command=None, env=None, resources=None, image_pull_policy=None, labels=None, ports=None, restart='Always')

Generate a pod definition.

Args:

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

Returns:

A kr8s.objects.Pod object.

Example:

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

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

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

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

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

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

Bases: kr8s._objects.PodTemplate

A Kubernetes PodTemplate.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'podtemplates'

kind = 'PodTemplate'

plural = 'podtemplates'

singular = 'podtemplate'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.ReplicationController

A Kubernetes ReplicationController.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'replicationcontrollers'

kind = 'ReplicationController'

plural = 'replicationcontrollers'

singular = 'replicationcontroller'

namespaced = True

scalable = True

scalable_spec = 'replicas'

async ready()

Check if the deployment is ready.

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.ResourceQuota

A Kubernetes ResourceQuota.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'resourcequotas'

kind = 'ResourceQuota'

plural = 'resourcequotas'

singular = 'resourcequota'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.Secret

A Kubernetes Secret.

property data: box.Box

Data of the Secret.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'secrets'

kind = 'Secret'

plural = 'secrets'

singular = 'secret'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.Service

A Kubernetes Service.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'services'

kind = 'Service'

plural = 'services'

singular = 'service'

namespaced = True

scalable = False

scalable_spec = 'replicas'

async proxy_http_request(method: str, path: str, port: int | None = None, **kwargs: Any) → httpx.Response

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

Args:

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

Service’s spec will be used.

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

async async_proxy_http_request(method: str, path: str, port: int | None = None, **kwargs: Any) → httpx.Response

async proxy_http_get(path: str, port: int | None = None, **kwargs) → httpx.Response

async proxy_http_post(path: str, port: int | None = None, **kwargs) → None

async proxy_http_put(path: str, port: int | None = None, **kwargs) → httpx.Response

async proxy_http_delete(path: str, port: int | None = None, **kwargs) → httpx.Response

async ready_pods() → List[Pod]

Return a list of ready Pods for this Service.

async async_ready_pods() → List[Pod]

Return a list of ready Pods for this Service.

async ready() → bool

Check if the service is ready.

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

Port forward a service.

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

Example:

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

Context manager:

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

Explict start/stop:

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

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.ServiceAccount

A Kubernetes ServiceAccount.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'v1'

endpoint = 'serviceaccounts'

kind = 'ServiceAccount'

plural = 'serviceaccounts'

singular = 'serviceaccount'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.ControllerRevision

A Kubernetes ControllerRevision.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'apps/v1'

endpoint = 'controllerrevisions'

kind = 'ControllerRevision'

plural = 'controllerrevisions'

singular = 'controllerrevision'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.DaemonSet

A Kubernetes DaemonSet.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'apps/v1'

endpoint = 'daemonsets'

kind = 'DaemonSet'

plural = 'daemonsets'

singular = 'daemonset'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.Deployment

A Kubernetes Deployment.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'apps/v1'

endpoint = 'deployments'

kind = 'Deployment'

plural = 'deployments'

singular = 'deployment'

namespaced = True

scalable = True

scalable_spec = 'replicas'

async pods() → List[Pod]

Return a list of Pods for this Deployment.

async ready()

Check if the deployment is ready.

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.ReplicaSet

A Kubernetes ReplicaSet.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'apps/v1'

endpoint = 'replicasets'

kind = 'ReplicaSet'

plural = 'replicasets'

singular = 'replicaset'

namespaced = True

scalable = True

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.StatefulSet

A Kubernetes StatefulSet.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'apps/v1'

endpoint = 'statefulsets'

kind = 'StatefulSet'

plural = 'statefulsets'

singular = 'statefulset'

namespaced = True

scalable = True

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.HorizontalPodAutoscaler

A Kubernetes HorizontalPodAutoscaler.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'autoscaling/v2'

endpoint = 'horizontalpodautoscalers'

kind = 'HorizontalPodAutoscaler'

plural = 'horizontalpodautoscalers'

singular = 'horizontalpodautoscaler'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.CronJob

A Kubernetes CronJob.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'batch/v1'

endpoint = 'cronjobs'

kind = 'CronJob'

plural = 'cronjobs'

singular = 'cronjob'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.Job

A Kubernetes Job.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'batch/v1'

endpoint = 'jobs'

kind = 'Job'

plural = 'jobs'

singular = 'job'

namespaced = True

scalable = True

scalable_spec = 'parallelism'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.Ingress

A Kubernetes Ingress.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'networking.k8s.io/v1'

endpoint = 'ingresses'

kind = 'Ingress'

plural = 'ingresses'

singular = 'ingress'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.IngressClass

A Kubernetes IngressClass.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'networking.k8s.io/v1'

endpoint = 'ingressclasses'

kind = 'IngressClass'

plural = 'ingressclasses'

singular = 'ingressclass'

namespaced = False

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.NetworkPolicy

A Kubernetes NetworkPolicy.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'networking.k8s.io/v1'

endpoint = 'networkpolicies'

kind = 'NetworkPolicy'

plural = 'networkpolicies'

singular = 'networkpolicy'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.PodDisruptionBudget

A Kubernetes PodDisruptionBudget.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'policy/v1'

endpoint = 'poddisruptionbudgets'

kind = 'PodDisruptionBudget'

plural = 'poddisruptionbudgets'

singular = 'poddisruptionbudget'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.ClusterRoleBinding

A Kubernetes ClusterRoleBinding.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'rbac.authorization.k8s.io/v1'

endpoint = 'clusterrolebindings'

kind = 'ClusterRoleBinding'

plural = 'clusterrolebindings'

singular = 'clusterrolebinding'

namespaced = False

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.ClusterRole

A Kubernetes ClusterRole.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'rbac.authorization.k8s.io/v1'

endpoint = 'clusterroles'

kind = 'ClusterRole'

plural = 'clusterroles'

singular = 'clusterrole'

namespaced = False

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.RoleBinding

A Kubernetes RoleBinding.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'rbac.authorization.k8s.io/v1'

endpoint = 'rolebindings'

kind = 'RoleBinding'

plural = 'rolebindings'

singular = 'rolebinding'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.Role

A Kubernetes Role.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'rbac.authorization.k8s.io/v1'

endpoint = 'roles'

kind = 'Role'

plural = 'roles'

singular = 'role'

namespaced = True

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.CustomResourceDefinition

A Kubernetes CustomResourceDefinition.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'apiextensions.k8s.io/v1'

endpoint = 'customresourcedefinitions'

kind = 'CustomResourceDefinition'

plural = 'customresourcedefinitions'

singular = 'customresourcedefinition'

namespaced = False

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

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

Bases: kr8s._objects.Table

A Kubernetes Table.

property rows: List[Dict]

Table rows.

property column_definitions: List[Dict]

Table column definitions.

property raw: str

Raw object returned from the Kubernetes API.

property name: str

Name of the Kubernetes resource.

property namespace: str

Namespace of the Kubernetes resource.

property metadata: box.Box

Metadata of the Kubernetes resource.

property spec: box.Box

Spec of the Kubernetes resource.

property status: box.Box

Status of the Kubernetes resource.

property labels: box.Box

Labels of the Kubernetes resource.

property annotations: box.Box

Annotations of the Kubernetes resource.

property replicas: int

Replicas of the Kubernetes resource.

version = 'meta.k8s.io/v1'

endpoint = 'tables'

kind = 'Table'

plural = 'tables'

singular = 'table'

namespaced = False

scalable = False

scalable_spec = 'replicas'

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

Get a Kubernetes resource by name or via selectors.

async exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async async_exists(ensure=False) → bool

Check if this object exists in Kubernetes.

async create() → None

Create this object in Kubernetes.

async delete(propagation_policy: str = None) → None

Delete this object from Kubernetes.

async refresh() → None

Refresh this object from Kubernetes.

async async_refresh() → None

Refresh this object from Kubernetes.

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

Patch this object in Kubernetes.

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

Patch this object in Kubernetes.

async scale(replicas: int = None) → None

Scale this object in Kubernetes.

async async_watch()

Watch this object in Kubernetes.

async watch()

Watch this object in Kubernetes.

async classmethod list(**kwargs) → List[APIObjectType]

List objects in Kubernetes.

Args:

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

Returns:

A list of objects.

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

Wait for conditions to be met.

Args:

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

Example:

Wait for a Pod to be ready.

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

Wait for a Job to either succeed or fail.

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

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

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

Note:

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

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

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

Annotate this object in Kubernetes.

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

Add labels to this object in Kubernetes.

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

Args:

labels:

A dictionary of labels to set.

**kwargs:

Labels to set.

Example:

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

keys() → list

Return the keys of this object.

async set_owner(owner: APIObject) → None

Set the owner reference of this object.

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

Args:

owner: The owner object to set a reference to.

Example:

>>> from kr8s.objects import Deployment, Pod
>>> deployment = Deployment.get("my-deployment")
>>> pod = Pod.get("my-pod")
>>> pod.set_owner(deployment)

async async_set_owner(owner: APIObject) → None

Set the owner of this object.

async adopt(child: APIObject) → None

Adopt this object.

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

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

Args:

child: The child object to adopt.

Example:

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

to_dict() → dict

Return a dictionary representation of this object.

to_lightkube() → Any

Return a lightkube representation of this object.

to_pykube(api) → Any

Return a pykube representation of this object.

Args:

api: A pykube API object.

Example:

>>> from kr8s.objects import Deployment
>>> deployment = Deployment.get("my-deployment")
>>> # Create a pykube API object
>>> from pykube import HTTPClient
>>> api = HTTPClient()
>>> pykube_deployment = deployment.to_pykube(api)

abstract classmethod gen(*args, **kwargs)

kr8s.objects.object_from_name_type

kr8s.objects.objects_from_files