kr8s.asyncio.objects ==================== .. py:module:: kr8s.asyncio.objects .. autoapi-nested-parse:: 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 ------- .. autoapisummary:: kr8s.asyncio.objects.APIObject kr8s.asyncio.objects.Binding kr8s.asyncio.objects.ClusterRole kr8s.asyncio.objects.ClusterRoleBinding kr8s.asyncio.objects.ComponentStatus kr8s.asyncio.objects.ConfigMap kr8s.asyncio.objects.ControllerRevision kr8s.asyncio.objects.CronJob kr8s.asyncio.objects.CustomResourceDefinition kr8s.asyncio.objects.DaemonSet kr8s.asyncio.objects.Deployment kr8s.asyncio.objects.Endpoints kr8s.asyncio.objects.Event kr8s.asyncio.objects.HorizontalPodAutoscaler kr8s.asyncio.objects.Ingress kr8s.asyncio.objects.IngressClass kr8s.asyncio.objects.IPAddress kr8s.asyncio.objects.Job kr8s.asyncio.objects.LimitRange kr8s.asyncio.objects.Namespace kr8s.asyncio.objects.NetworkPolicy kr8s.asyncio.objects.Node kr8s.asyncio.objects.PersistentVolume kr8s.asyncio.objects.PersistentVolumeClaim kr8s.asyncio.objects.Pod kr8s.asyncio.objects.PodDisruptionBudget kr8s.asyncio.objects.PodTemplate kr8s.asyncio.objects.ReplicaSet kr8s.asyncio.objects.ReplicationController kr8s.asyncio.objects.ResourceQuota kr8s.asyncio.objects.Role kr8s.asyncio.objects.RoleBinding kr8s.asyncio.objects.Secret kr8s.asyncio.objects.Service kr8s.asyncio.objects.ServiceAccount kr8s.asyncio.objects.ServiceCIDR kr8s.asyncio.objects.StatefulSet kr8s.asyncio.objects.Table Functions --------- .. autoapisummary:: kr8s.asyncio.objects.get_class kr8s.asyncio.objects.new_class kr8s.asyncio.objects.object_from_name_type kr8s.asyncio.objects.object_from_spec kr8s.asyncio.objects.objects_from_files Module Contents --------------- .. py:class:: APIObject(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Base class for Kubernetes objects. .. py:attribute:: version :type: str .. py:attribute:: endpoint :type: str .. py:attribute:: kind :type: str .. py:attribute:: plural :type: str .. py:attribute:: singular :type: str .. py:attribute:: namespaced :type: bool :value: False .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: Binding(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes Binding. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'bindings' .. py:attribute:: kind :value: 'Binding' .. py:attribute:: plural :value: 'bindings' .. py:attribute:: singular :value: 'binding' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: ClusterRole(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes ClusterRole. .. py:attribute:: version :value: 'rbac.authorization.k8s.io/v1' .. py:attribute:: endpoint :value: 'clusterroles' .. py:attribute:: kind :value: 'ClusterRole' .. py:attribute:: plural :value: 'clusterroles' .. py:attribute:: singular :value: 'clusterrole' .. py:attribute:: namespaced :value: False .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: ClusterRoleBinding(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes ClusterRoleBinding. .. py:attribute:: version :value: 'rbac.authorization.k8s.io/v1' .. py:attribute:: endpoint :value: 'clusterrolebindings' .. py:attribute:: kind :value: 'ClusterRoleBinding' .. py:attribute:: plural :value: 'clusterrolebindings' .. py:attribute:: singular :value: 'clusterrolebinding' .. py:attribute:: namespaced :value: False .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: ComponentStatus(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes ComponentStatus. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'componentstatuses' .. py:attribute:: kind :value: 'ComponentStatus' .. py:attribute:: plural :value: 'componentstatuses' .. py:attribute:: singular :value: 'componentstatus' .. py:attribute:: namespaced :value: False .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: ConfigMap(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes ConfigMap. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'configmaps' .. py:attribute:: kind :value: 'ConfigMap' .. py:attribute:: plural :value: 'configmaps' .. py:attribute:: singular :value: 'configmap' .. py:attribute:: namespaced :value: True .. py:property:: data :type: box.Box Data of the ConfigMap. .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: ControllerRevision(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes ControllerRevision. .. py:attribute:: version :value: 'apps/v1' .. py:attribute:: endpoint :value: 'controllerrevisions' .. py:attribute:: kind :value: 'ControllerRevision' .. py:attribute:: plural :value: 'controllerrevisions' .. py:attribute:: singular :value: 'controllerrevision' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: CronJob(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes CronJob. .. py:attribute:: version :value: 'batch/v1' .. py:attribute:: endpoint :value: 'cronjobs' .. py:attribute:: kind :value: 'CronJob' .. py:attribute:: plural :value: 'cronjobs' .. py:attribute:: singular :value: 'cronjob' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: CustomResourceDefinition(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes CustomResourceDefinition. .. py:attribute:: version :value: 'apiextensions.k8s.io/v1' .. py:attribute:: endpoint :value: 'customresourcedefinitions' .. py:attribute:: kind :value: 'CustomResourceDefinition' .. py:attribute:: plural :value: 'customresourcedefinitions' .. py:attribute:: singular :value: 'customresourcedefinition' .. py:attribute:: namespaced :value: False .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: DaemonSet(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes DaemonSet. .. py:attribute:: version :value: 'apps/v1' .. py:attribute:: endpoint :value: 'daemonsets' .. py:attribute:: kind :value: 'DaemonSet' .. py:attribute:: plural :value: 'daemonsets' .. py:attribute:: singular :value: 'daemonset' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: Deployment(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes Deployment. .. py:attribute:: version :value: 'apps/v1' .. py:attribute:: endpoint :value: 'deployments' .. py:attribute:: kind :value: 'Deployment' .. py:attribute:: plural :value: 'deployments' .. py:attribute:: singular :value: 'deployment' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :value: True .. py:method:: ready_pods() -> list[Pod] :async: Return a list of Pods for this Deployment. .. py:method:: pods() -> list[Pod] :async: Return a list of Pods for this Deployment. .. py:method:: ready() :async: Check if the deployment is ready. .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: Endpoints(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes Endpoints. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'endpoints' .. py:attribute:: kind :value: 'Endpoints' .. py:attribute:: plural :value: 'endpoints' .. py:attribute:: singular :value: 'endpoint' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: Event(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes Event. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'events' .. py:attribute:: kind :value: 'Event' .. py:attribute:: plural :value: 'events' .. py:attribute:: singular :value: 'event' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: HorizontalPodAutoscaler(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes HorizontalPodAutoscaler. .. py:attribute:: version :value: 'autoscaling/v2' .. py:attribute:: endpoint :value: 'horizontalpodautoscalers' .. py:attribute:: kind :value: 'HorizontalPodAutoscaler' .. py:attribute:: plural :value: 'horizontalpodautoscalers' .. py:attribute:: singular :value: 'horizontalpodautoscaler' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: Ingress(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes Ingress. .. py:attribute:: version :value: 'networking.k8s.io/v1' .. py:attribute:: endpoint :value: 'ingresses' .. py:attribute:: kind :value: 'Ingress' .. py:attribute:: plural :value: 'ingresses' .. py:attribute:: singular :value: 'ingress' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: IngressClass(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes IngressClass. .. py:attribute:: version :value: 'networking.k8s.io/v1' .. py:attribute:: endpoint :value: 'ingressclasses' .. py:attribute:: kind :value: 'IngressClass' .. py:attribute:: plural :value: 'ingressclasses' .. py:attribute:: singular :value: 'ingressclass' .. py:attribute:: namespaced :value: False .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: IPAddress(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes IPAddress. .. py:attribute:: version :value: 'networking.k8s.io/v1' .. py:attribute:: endpoint :value: 'ipaddresses' .. py:attribute:: kind :value: 'IPAddress' .. py:attribute:: plural :value: 'ipaddresses' .. py:attribute:: singular :value: 'ipaddress' .. py:attribute:: namespaced :value: False .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: Job(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes Job. .. py:attribute:: version :value: 'batch/v1' .. py:attribute:: endpoint :value: 'jobs' .. py:attribute:: kind :value: 'Job' .. py:attribute:: plural :value: 'jobs' .. py:attribute:: singular :value: 'job' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :value: True .. py:attribute:: scalable_spec :value: 'parallelism' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: LimitRange(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes LimitRange. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'limitranges' .. py:attribute:: kind :value: 'LimitRange' .. py:attribute:: plural :value: 'limitranges' .. py:attribute:: singular :value: 'limitrange' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: Namespace(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes Namespace. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'namespaces' .. py:attribute:: kind :value: 'Namespace' .. py:attribute:: plural :value: 'namespaces' .. py:attribute:: singular :value: 'namespace' .. py:attribute:: namespaced :value: False .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: NetworkPolicy(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes NetworkPolicy. .. py:attribute:: version :value: 'networking.k8s.io/v1' .. py:attribute:: endpoint :value: 'networkpolicies' .. py:attribute:: kind :value: 'NetworkPolicy' .. py:attribute:: plural :value: 'networkpolicies' .. py:attribute:: singular :value: 'networkpolicy' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: Node(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes Node. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'nodes' .. py:attribute:: kind :value: 'Node' .. py:attribute:: plural :value: 'nodes' .. py:attribute:: singular :value: 'node' .. py:attribute:: namespaced :value: False .. py:property:: unschedulable .. py:method:: cordon() -> None :async: Cordon the node. This will mark the node as unschedulable. .. py:method:: uncordon() -> None :async: Uncordon the node. This will mark the node as schedulable. .. py:method:: taint(key: str, value: str, *, effect: str) -> None :async: Taint a node. .. py:property:: taints :type: box.Box Labels of the Kubernetes resource. .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: PersistentVolume(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes PersistentVolume. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'persistentvolumes' .. py:attribute:: kind :value: 'PersistentVolume' .. py:attribute:: plural :value: 'persistentvolumes' .. py:attribute:: singular :value: 'persistentvolume' .. py:attribute:: namespaced :value: False .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: PersistentVolumeClaim(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes PersistentVolumeClaim. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'persistentvolumeclaims' .. py:attribute:: kind :value: 'PersistentVolumeClaim' .. py:attribute:: plural :value: 'persistentvolumeclaims' .. py:attribute:: singular :value: 'persistentvolumeclaim' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: Pod(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes Pod. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'pods' .. py:attribute:: kind :value: 'Pod' .. py:attribute:: plural :value: 'pods' .. py:attribute:: singular :value: 'pod' .. py:attribute:: namespaced :value: True .. py:method:: ready() -> bool :async: Check if the pod is ready. .. py:method:: logs(container=None, pretty=None, previous=False, since_seconds=None, since_time=None, timestamps=False, tail_lines=None, limit_bytes=None, follow=False, timeout=3600) -> collections.abc.AsyncGenerator[str] :async: .. py:method:: portforward(remote_port: int, local_port: kr8s.portforward.LocalPortType = 'match', address: list[str] | str = '127.0.0.1') -> kr8s.asyncio.portforward.PortForward Port forward a pod. Returns an instance of :class:`kr8s.portforward.PortForward` for this Pod. Args: remote_port: The port on the Pod to forward to. local_port: The local port to listen on. Defaults to ``"match"``, which will match the ``remote_port``. Set to ``"auto"`` or ``None`` to find an available high port. Set to an ``int`` to specify a specific port. address: List of addresses or address to listen on. Defaults to ["127.0.0.1"], will listen only on 127.0.0.1. Example: This can be used as a an async context manager or with explicit start/stop methods. Context manager: >>> async with pod.portforward(8888) as port: ... print(f"Forwarding to port {port}") ... # Do something with port 8888 Explict start/stop: >>> pf = pod.portforward(8888) >>> await pf.start() >>> print(f"Forwarding to port {pf.local_port}") >>> # Do something with port 8888 >>> await pf.stop() Explict bind address: >>> async with pod.PortForward(8888, address=["127.0.0.1", "10.20.0.1"]) as port: ... print(f"Forwarding to port {port}") ... # Do something with port 8888 on the Pod, port will be bind to 127.0.0.1 and 10.20.0.1 .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) :async: Run a command in a container and wait until it completes. Behaves like :func:`subprocess.run`. Args: command: Command to execute. container: Container to execute the command in. stdin: If set, read stdin to the container. stdout: If set, write stdout to the provided writable stream object. stderr: If set, write stderr to the provided writable stream object. check: If True, raise an exception if the command fails. capture_output: If True, store stdout and stderr from the container in an attribute. Returns: A :class:`kr8s._exec.CompletedExec` object. Example: >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> ex = await pod.exec(["ls", "-l"]) >>> print(ex.stdout) >>> print(ex.stderr) .. py:method:: gen(*, name=None, generate_name=None, image=None, namespace=None, annotations=None, command=None, env=None, resources=None, image_pull_policy=None, labels=None, ports=None, restart='Always') :classmethod: Generate a pod definition. Args: name (str): The name of the pod. generate_name (str): Template for generating the name of the pod. namespace (str): The namespace of the pod. image (str): The image to use. annotations (dict): Annotations to add to the pod. command (list): Command to run in the container. env (dict): Environment variables to set in the container. resources (dict): Resources to set in the container. image_pull_policy (str): Image pull policy to use. labels (dict): Labels to add to the pod. ports (list|int): Ports to expose. restart (str): Restart policy to use. Returns: A :class:`kr8s.objects.Pod` object. Example: >>> from kr8s.objects import Pod >>> pod = Pod.gen(name="my-pod", image="my-image") >>> pod.create() Create an nginx Pod that exposes port 80. >>> from kr8s.objects import Pod >>> pod = Pod.gen(name="nginx", image="nginx:latest", ports=[80]) >>> pod.create() Create an wordpress Pod that exposes port 80. >>> from kr8s.objects import Pod >>> pod = Pod.gen(name="wordpress", image="wordpress:latest", ports=[{"containerPort": 80}]) >>> pod.create() Create a Pod that requires a GPU >>> from kr8s.objects import Pod >>> pod = Pod.gen(name="cuda-vectoradd", ... image="nvidia/samples:vectoradd-cuda11.6.0-ubuntu18.04", ... resources={"limits": {"nvidia.com/gpu": 1}}) .. py:method:: tolerate(key: str, *, operator: str, effect: str, value: str | None = None, toleration_seconds: int | None = None) :async: Add a toleration to the Pod. Args: key (str): Key of the taint to tolerate. operator (str): Toleration operator. effect (str): Specifies the effect of the taint to tolerate. value (str): Value of taint to tolerate. toleration_seconds (str): Toleration seconds. .. py:property:: tolerations :type: box.Box Tolerations for Pod. .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: PodDisruptionBudget(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes PodDisruptionBudget. .. py:attribute:: version :value: 'policy/v1' .. py:attribute:: endpoint :value: 'poddisruptionbudgets' .. py:attribute:: kind :value: 'PodDisruptionBudget' .. py:attribute:: plural :value: 'poddisruptionbudgets' .. py:attribute:: singular :value: 'poddisruptionbudget' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: PodTemplate(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes PodTemplate. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'podtemplates' .. py:attribute:: kind :value: 'PodTemplate' .. py:attribute:: plural :value: 'podtemplates' .. py:attribute:: singular :value: 'podtemplate' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: ReplicaSet(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes ReplicaSet. .. py:attribute:: version :value: 'apps/v1' .. py:attribute:: endpoint :value: 'replicasets' .. py:attribute:: kind :value: 'ReplicaSet' .. py:attribute:: plural :value: 'replicasets' .. py:attribute:: singular :value: 'replicaset' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :value: True .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: ReplicationController(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes ReplicationController. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'replicationcontrollers' .. py:attribute:: kind :value: 'ReplicationController' .. py:attribute:: plural :value: 'replicationcontrollers' .. py:attribute:: singular :value: 'replicationcontroller' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :value: True .. py:method:: ready() :async: Check if the deployment is ready. .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: ResourceQuota(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes ResourceQuota. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'resourcequotas' .. py:attribute:: kind :value: 'ResourceQuota' .. py:attribute:: plural :value: 'resourcequotas' .. py:attribute:: singular :value: 'resourcequota' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: Role(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes Role. .. py:attribute:: version :value: 'rbac.authorization.k8s.io/v1' .. py:attribute:: endpoint :value: 'roles' .. py:attribute:: kind :value: 'Role' .. py:attribute:: plural :value: 'roles' .. py:attribute:: singular :value: 'role' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: RoleBinding(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes RoleBinding. .. py:attribute:: version :value: 'rbac.authorization.k8s.io/v1' .. py:attribute:: endpoint :value: 'rolebindings' .. py:attribute:: kind :value: 'RoleBinding' .. py:attribute:: plural :value: 'rolebindings' .. py:attribute:: singular :value: 'rolebinding' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: Secret(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes Secret. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'secrets' .. py:attribute:: kind :value: 'Secret' .. py:attribute:: plural :value: 'secrets' .. py:attribute:: singular :value: 'secret' .. py:attribute:: namespaced :value: True .. py:property:: data :type: box.Box Data of the Secret. .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: Service(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes Service. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'services' .. py:attribute:: kind :value: 'Service' .. py:attribute:: plural :value: 'services' .. py:attribute:: singular :value: 'service' .. py:attribute:: namespaced :value: True .. py:method:: proxy_http_request(method: str, path: str, port: int | None = None, **kwargs: Any) -> httpx.Response :async: Issue a HTTP request with specific HTTP method to proxy of a Service. Args: method: HTTP method to use. path: Path to proxy. port: Port to proxy to. If not specified, the first port in the Service's spec will be used. **kwargs: Additional keyword arguments to pass to the API call. .. py:method:: proxy_http_get(path: str, port: int | None = None, **kwargs) -> httpx.Response :async: .. py:method:: proxy_http_post(path: str, port: int | None = None, **kwargs) -> None :async: .. py:method:: proxy_http_put(path: str, port: int | None = None, **kwargs) -> httpx.Response :async: .. py:method:: proxy_http_delete(path: str, port: int | None = None, **kwargs) -> httpx.Response :async: .. py:method:: ready_pods() -> list[Pod] :async: Return a list of ready Pods for this Service. .. py:method:: ready() -> bool :async: Check if the service is ready. .. py:method:: portforward(remote_port: int, local_port: kr8s.portforward.LocalPortType = 'match', address: str | list[str] = '127.0.0.1') -> kr8s.asyncio.portforward.PortForward Port forward a service. Returns an instance of :class:`kr8s.portforward.PortForward` for this Service. Args: remote_port: The port on the Pod to forward to. local_port: The local port to listen on. Defaults to ``"match"``, which will match the ``remote_port``. Set to ``"auto"`` or ``None`` to find an available high port. Set to an ``int`` to specify a specific port. address: List of addresses or address to listen on. Defaults to ["127.0.0.1"], will listen only on 127.0.0.1. Example: This can be used as a an async context manager or with explicit start/stop methods. Context manager: >>> async with service.portforward(8888) as port: ... print(f"Forwarding to port {port}") ... # Do something with port 8888 Explict start/stop: >>> pf = service.portforward(8888) >>> await pf.start() >>> print(f"Forwarding to port {pf.local_port}") >>> # Do something with port 8888 >>> await pf.stop() .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: ServiceAccount(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes ServiceAccount. .. py:attribute:: version :value: 'v1' .. py:attribute:: endpoint :value: 'serviceaccounts' .. py:attribute:: kind :value: 'ServiceAccount' .. py:attribute:: plural :value: 'serviceaccounts' .. py:attribute:: singular :value: 'serviceaccount' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: ServiceCIDR(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes ServiceCIDR. .. py:attribute:: version :value: 'networking.k8s.io/v1' .. py:attribute:: endpoint :value: 'servicecidrs' .. py:attribute:: kind :value: 'ServiceCIDR' .. py:attribute:: plural :value: 'servicecidrs' .. py:attribute:: singular :value: 'servicecidr' .. py:attribute:: namespaced :value: False .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: StatefulSet(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes StatefulSet. .. py:attribute:: version :value: 'apps/v1' .. py:attribute:: endpoint :value: 'statefulsets' .. py:attribute:: kind :value: 'StatefulSet' .. py:attribute:: plural :value: 'statefulsets' .. py:attribute:: singular :value: 'statefulset' .. py:attribute:: namespaced :value: True .. py:attribute:: scalable :value: True .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:class:: Table(resource: kr8s._types.SpecType, namespace: str | None = None, api: kr8s._api.Api | None = None) Bases: :py:obj:`APIObject` A Kubernetes Table. .. py:attribute:: version :value: 'meta.k8s.io/v1' .. py:attribute:: endpoint :value: 'tables' .. py:attribute:: kind :value: 'Table' .. py:attribute:: plural :value: 'tables' .. py:attribute:: singular :value: 'table' .. py:attribute:: namespaced :value: False .. py:property:: rows :type: list[dict] Table rows. .. py:property:: column_definitions :type: list[dict] Table column definitions. .. py:attribute:: scalable :type: bool :value: False .. py:attribute:: scalable_spec :type: str :value: 'replicas' .. py:property:: api .. py:property:: raw :type: box.Box Raw object returned from the Kubernetes API. .. py:property:: name :type: str Name of the Kubernetes resource. .. py:property:: namespace :type: str | None Namespace of the Kubernetes resource. .. py:property:: metadata :type: box.Box Metadata of the Kubernetes resource. .. py:property:: spec :type: box.Box Spec of the Kubernetes resource. .. py:property:: status :type: box.Box Status of the Kubernetes resource. .. py:property:: labels :type: box.Box Labels of the Kubernetes resource. .. py:property:: annotations :type: box.Box Annotations of the Kubernetes resource. .. py:property:: replicas :type: int Replicas of the Kubernetes resource. .. py:method:: 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 :classmethod: :async: Get a Kubernetes resource by name or via selectors. .. py:method:: exists(ensure=False) -> bool :async: Check if this object exists in Kubernetes. .. py:method:: create() -> None :async: Create this object in Kubernetes. .. py:method:: delete(propagation_policy: str | None = None, grace_period: int | None = None, force: bool = False) -> None :async: 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) .. py:method:: refresh() -> None :async: Refresh this object from Kubernetes. .. py:method:: patch(patch, *, subresource=None, type=None) -> None :async: Patch this object in Kubernetes. .. py:method:: scale(replicas: int | None = None) -> None :async: Scale this object in Kubernetes. .. py:method:: exec(command: list[str], *, container: str | None = None, stdin: str | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, check: bool = True, capture_output: bool = True) -> kr8s._exec.Exec :async: Execute a command in this object. .. py:method:: watch() -> collections.abc.AsyncGenerator[tuple[str, typing_extensions.Self]] :async: Watch this object in Kubernetes. .. py:method:: wait(conditions: list[str] | str, mode: Literal['any', 'all'] = 'any', timeout: int | None = None) :async: Wait for conditions to be met. Args: conditions: A list of conditions to wait for. mode: Match any condition with "any" or all conditions with "all". Defaults to "any". timeout: Timeout in seconds. Example: Wait for a Pod to be ready. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait("condition=Ready") Wait for a Job to either succeed or fail. >>> from kr8s.objects import Job >>> job = Job.get("my-jod") >>> job.wait(["condition=Complete", "condition=Failed"]) Wait for a Pod to be initialized and ready to start containers. >>> from kr8s.objects import Pod >>> pod = Pod.get("my-pod") >>> pod.wait(["condition=Initialized", "condition=PodReadyToStartContainers"], mode="all") Note: As of the current Kubertenetes release when this function was written (1.29) kubectl doesn't support multiple conditions. There is a PR to implement this but it hasn't been merged yet https://github.com/kubernetes/kubernetes/pull/119205. Given that ``for`` is a reserved word anyway we can't exactly match the kubectl API so we use ``condition`` in combination with a ``mode`` instead. .. py:method:: annotate(annotations: dict | None = None, **kwargs) -> None :async: Annotate this object in Kubernetes. .. py:method:: label(*labels: dict | str, **kwargs) -> None :async: 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 or string to remove. **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") >>> # You can also remove a label by passing it's name with a `-` on the end >>> deployment.label("app-") .. py:method:: remove_label(*labels: str) -> None :async: Remove labels from this object in Kubernetes. Labels can be passed as a list of strings. Args: labels: A list of labels to remove. Example: >>> from kr8s.objects import Deployment >>> deployment = Deployment.get("my-deployment") >>> deployment.remove_label("app") .. py:method:: keys() -> list Return the keys of this object. .. py:method:: set_owner(owner: APIObject) -> None :async: 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) .. py:method:: adopt(child: APIObject) -> None :async: 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) .. py:method:: to_dict() -> dict Return a dictionary representation of this object. .. py:method:: to_yaml() -> str Return a YAML representation of this object. .. py:method:: to_lightkube() -> Any Return a lightkube representation of this object. .. py:method:: 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) .. py:method:: pprint(use_rich: bool = True, theme: str = 'ansi_dark') -> None Pretty print this object to stdout. Prints the object as YAML (using ``rich`` for syntax highlighting if available). Args: use_rich: Use ``rich`` to pretty print. If ``rich` is not installed this will be ignored. theme: The ``pygments`` theme for ``rich`` to use. Defaults to "ansi_dark" to use default terminal colors. .. py:method:: gen(*args, **kwargs) :classmethod: :abstractmethod: .. py:method:: list(**kwargs) -> collections.abc.AsyncGenerator[typing_extensions.Self] :classmethod: :async: List objects in Kubernetes. Args: api: An optional API object to use. **kwargs: Keyword arguments to pass to :func:`kr8s.get`. Returns: A list of objects. .. py:function:: get_class(kind: str, version: str | None = None, _asyncio: bool = True) -> type[APIObject] Get an APIObject subclass by kind and version. Args: kind: The Kubernetes resource kind. version: The Kubernetes API group/version. Returns: An APIObject subclass. Raises: KeyError: If no object is registered for the given kind and version. .. py:function:: new_class(kind: str, version: str | None = None, asyncio: bool = True, namespaced=True, scalable: bool | None = None, scalable_spec: str | None = None, plural: str | None = None) -> type[APIObject] Create a new APIObject subclass. Args: kind: The Kubernetes resource kind. version: The Kubernetes API version. asyncio: Whether to use asyncio or not. namespaced: Whether the resource is namespaced or not. scalable: Whether the resource is scalable or not. scalable_spec: The name of the field to use for scaling. plural: The plural form of the resource. Returns: A new APIObject subclass. .. py:function:: object_from_name_type(name: str, namespace: str | None = None, api: kr8s._api.Api | None = None, _asyncio: bool = True) -> APIObject :async: Create an APIObject from a Kubernetes resource name. Args: name: A Kubernetes resource name. namespace: The namespace of the resource. api: An optional API instance to use. _asyncio: Whether to use asyncio or not. Returns: A corresponding APIObject subclass instance. Raises: ValueError: If the resource kind or API version is not supported. .. py:function:: object_from_spec(spec: dict, api: kr8s._api.Api | None = None, allow_unknown_type: bool = False, _asyncio: bool = True) -> APIObject Create an APIObject from a Kubernetes resource spec. Args: spec: A Kubernetes resource spec. api: An optional API instance to use. allow_unknown_type: Whether to allow unknown resource types. _asyncio: Whether to use asyncio or not. Returns: A corresponding APIObject subclass instance. Raises: ValueError: If the resource kind or API version is not supported. .. py:function:: objects_from_files(path: str | pathlib.Path, api: kr8s._api.Api | None = None, recursive: bool = False, _asyncio: bool = True) -> list[APIObject] :async: Create APIObjects from Kubernetes resource files. Args: path: A path to a Kubernetes resource file or directory of resource files. api: An optional API instance to use. recursive: Whether to recursively search for resource files in subdirectories. _asyncio: Whether to use asyncio or not. Returns: A list of APIObject subclass instances. Raises: ValueError: If the resource kind or API version is not supported.