kr8s.asyncio.objects¶
Objects to represent Kubernetes resources.
This module provides classes that represent Kubernetes resources. These classes are used to interact with resources in the Kubernetes API server.
Classes¶
Base class for Kubernetes objects. |
|
A Kubernetes Binding. |
|
A Kubernetes ClusterRole. |
|
A Kubernetes ClusterRoleBinding. |
|
A Kubernetes ComponentStatus. |
|
A Kubernetes ConfigMap. |
|
A Kubernetes ControllerRevision. |
|
A Kubernetes CronJob. |
|
A Kubernetes CustomResourceDefinition. |
|
A Kubernetes DaemonSet. |
|
A Kubernetes Deployment. |
|
A Kubernetes Endpoints. |
|
A Kubernetes Event. |
|
A Kubernetes HorizontalPodAutoscaler. |
|
A Kubernetes Ingress. |
|
A Kubernetes IngressClass. |
|
A Kubernetes Job. |
|
A Kubernetes LimitRange. |
|
A Kubernetes Namespace. |
|
A Kubernetes NetworkPolicy. |
|
A Kubernetes Node. |
|
A Kubernetes PersistentVolume. |
|
A Kubernetes PersistentVolumeClaim. |
|
A Kubernetes Pod. |
|
A Kubernetes PodDisruptionBudget. |
|
A Kubernetes PodTemplate. |
|
A Kubernetes ReplicaSet. |
|
A Kubernetes ReplicationController. |
|
A Kubernetes ResourceQuota. |
|
A Kubernetes Role. |
|
A Kubernetes RoleBinding. |
|
A Kubernetes Secret. |
|
A Kubernetes Service. |
|
A Kubernetes ServiceAccount. |
|
A Kubernetes StatefulSet. |
|
A Kubernetes Table. |
Functions¶
|
Get an APIObject subclass by kind and version. |
|
Create a new APIObject subclass. |
|
Create an APIObject from a Kubernetes resource name. |
|
Create an APIObject from a Kubernetes resource spec. |
|
Create APIObjects from Kubernetes resource files. |
Module Contents¶
- class kr8s.asyncio.objects.APIObject(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Base class for Kubernetes objects.
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.Binding(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes Binding.
- version = 'v1'¶
- endpoint = 'bindings'¶
- kind = 'Binding'¶
- plural = 'bindings'¶
- singular = 'binding'¶
- namespaced = True¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.ClusterRole(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes ClusterRole.
- version = 'rbac.authorization.k8s.io/v1'¶
- endpoint = 'clusterroles'¶
- kind = 'ClusterRole'¶
- plural = 'clusterroles'¶
- singular = 'clusterrole'¶
- namespaced = False¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.ClusterRoleBinding(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes ClusterRoleBinding.
- version = 'rbac.authorization.k8s.io/v1'¶
- endpoint = 'clusterrolebindings'¶
- kind = 'ClusterRoleBinding'¶
- plural = 'clusterrolebindings'¶
- singular = 'clusterrolebinding'¶
- namespaced = False¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.ComponentStatus(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes ComponentStatus.
- version = 'v1'¶
- endpoint = 'componentstatuses'¶
- kind = 'ComponentStatus'¶
- plural = 'componentstatuses'¶
- singular = 'componentstatus'¶
- namespaced = False¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.ConfigMap(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes ConfigMap.
- version = 'v1'¶
- endpoint = 'configmaps'¶
- kind = 'ConfigMap'¶
- plural = 'configmaps'¶
- singular = 'configmap'¶
- namespaced = True¶
- property data: box.Box¶
Data of the ConfigMap.
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.ControllerRevision(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes ControllerRevision.
- version = 'apps/v1'¶
- endpoint = 'controllerrevisions'¶
- kind = 'ControllerRevision'¶
- plural = 'controllerrevisions'¶
- singular = 'controllerrevision'¶
- namespaced = True¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.CronJob(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes CronJob.
- version = 'batch/v1'¶
- endpoint = 'cronjobs'¶
- kind = 'CronJob'¶
- plural = 'cronjobs'¶
- singular = 'cronjob'¶
- namespaced = True¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.CustomResourceDefinition(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes CustomResourceDefinition.
- version = 'apiextensions.k8s.io/v1'¶
- endpoint = 'customresourcedefinitions'¶
- kind = 'CustomResourceDefinition'¶
- plural = 'customresourcedefinitions'¶
- singular = 'customresourcedefinition'¶
- namespaced = False¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.DaemonSet(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes DaemonSet.
- version = 'apps/v1'¶
- endpoint = 'daemonsets'¶
- kind = 'DaemonSet'¶
- plural = 'daemonsets'¶
- singular = 'daemonset'¶
- namespaced = True¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.Deployment(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes Deployment.
- version = 'apps/v1'¶
- endpoint = 'deployments'¶
- kind = 'Deployment'¶
- plural = 'deployments'¶
- singular = 'deployment'¶
- namespaced = True¶
- scalable = True¶
- async ready()¶
Check if the deployment is ready.
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.Endpoints(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes Endpoints.
- version = 'v1'¶
- endpoint = 'endpoints'¶
- kind = 'Endpoints'¶
- plural = 'endpoints'¶
- singular = 'endpoint'¶
- namespaced = True¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.Event(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes Event.
- version = 'v1'¶
- endpoint = 'events'¶
- kind = 'Event'¶
- plural = 'events'¶
- singular = 'event'¶
- namespaced = True¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.HorizontalPodAutoscaler(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes HorizontalPodAutoscaler.
- version = 'autoscaling/v2'¶
- endpoint = 'horizontalpodautoscalers'¶
- kind = 'HorizontalPodAutoscaler'¶
- plural = 'horizontalpodautoscalers'¶
- singular = 'horizontalpodautoscaler'¶
- namespaced = True¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.Ingress(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes Ingress.
- version = 'networking.k8s.io/v1'¶
- endpoint = 'ingresses'¶
- kind = 'Ingress'¶
- plural = 'ingresses'¶
- singular = 'ingress'¶
- namespaced = True¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.IngressClass(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes IngressClass.
- version = 'networking.k8s.io/v1'¶
- endpoint = 'ingressclasses'¶
- kind = 'IngressClass'¶
- plural = 'ingressclasses'¶
- singular = 'ingressclass'¶
- namespaced = False¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.Job(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes Job.
- version = 'batch/v1'¶
- endpoint = 'jobs'¶
- kind = 'Job'¶
- plural = 'jobs'¶
- singular = 'job'¶
- namespaced = True¶
- scalable = True¶
- scalable_spec = 'parallelism'¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- property 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.LimitRange(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes LimitRange.
- version = 'v1'¶
- endpoint = 'limitranges'¶
- kind = 'LimitRange'¶
- plural = 'limitranges'¶
- singular = 'limitrange'¶
- namespaced = True¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.Namespace(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes Namespace.
- version = 'v1'¶
- endpoint = 'namespaces'¶
- kind = 'Namespace'¶
- plural = 'namespaces'¶
- singular = 'namespace'¶
- namespaced = False¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"])
Wait for a Pod to be initialized and ready to start containers.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all")
- Note:
As of the current Kubertenetes release when this function was written (1.29) kubectl doesn’t support multiple conditions. There is a PR to implement this but it hasn’t been merged yet https://github.com/kubernetes/kubernetes/pull/119205.
Given that
for
is a reserved word anyway we can’t exactly match the kubectl API so we usecondition
in combination with amode
instead.
- async annotate(annotations: dict | None = None, **kwargs) None ¶
Annotate this object in Kubernetes.
- async label(labels: dict | None = None, **kwargs) None ¶
Add labels to this object in Kubernetes.
Labels can be passed as a dictionary or as keyword arguments.
- Args:
- labels:
A dictionary of labels to set.
- **kwargs:
Labels to set.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Both of these are equivalent >>> deployment.label({"app": "my-app"}) >>> deployment.label(app="my-app")
- async set_owner(owner: APIObject) None ¶
Set the owner reference of this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
owner: The owner object to set a reference to.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> pod.set_owner(deployment)
- async adopt(child: APIObject) None ¶
Adopt this object.
This will set the owner reference of the child object to this object.
See https://kubernetes.io/docs/concepts/overview/working-with-objects/owners-dependents/
- Args:
child: The child object to adopt.
- Example:
>>> from kr8s.objects import Deployment, Pod >>> deployment = Deployment.get("my-deployment") >>> pod = Pod.get("my-pod") >>> deployment.adopt(pod)
- to_lightkube() Any ¶
Return a lightkube representation of this object.
- to_pykube(api) Any ¶
Return a pykube representation of this object.
- Args:
api: A pykube API object.
- Example:
>>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> # Create a pykube API object >>> from pykube import HTTPClient >>> api = HTTPClient() >>> pykube_deployment = deployment.to_pykube(api)
- classmethod gen(*args, **kwargs)¶
- Abstractmethod:
- classmethod list(**kwargs) collections.abc.AsyncGenerator[typing_extensions.Self] ¶
- Async:
List objects in Kubernetes.
- Args:
api: An optional API object to use. **kwargs: Keyword arguments to pass to
kr8s.get()
.- Returns:
A list of objects.
- class kr8s.asyncio.objects.NetworkPolicy(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None)¶
Bases:
APIObject
A Kubernetes NetworkPolicy.
- version = 'networking.k8s.io/v1'¶
- endpoint = 'networkpolicies'¶
- kind = 'NetworkPolicy'¶
- plural = 'networkpolicies'¶
- singular = 'networkpolicy'¶
- namespaced = True¶
- property api¶
- property raw: Any¶
Raw object returned from the Kubernetes API.
- 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.
- classmethod get(name: str | None = None, namespace: str | None = None, api: kr8s._api.Api | None = None, label_selector: str | dict[str, str] | None = None, field_selector: str | dict[str, str] | None = None, timeout: int = 2, **kwargs) typing_extensions.Self ¶
- Async:
Get a Kubernetes resource by name or via selectors.
- async delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) None ¶
Delete this object from Kubernetes.
- Args:
propagation_policy: The deletion propagation policy. grace_period: The grace period for deletion. force: Force deletion. (Setting to
True
is equivelaent to setting grace_period to 0)
- async watch() collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] ¶
Watch this object in Kubernetes.
- async wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None)¶
Wait for conditions to be met.
- Args:
conditions: A list of conditions to wait for. mode: Match any condition with “any” or all conditions with “all”. Defaults to “any”. timeout: Timeout in seconds.
- Example:
Wait for a Pod to be ready.
>>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready")
Wait for a Job to either succeed or fail.
>>> from kr8s.objects import Job >>>