kr8s
====

.. py:module:: kr8s

.. autoapi-nested-parse::

   This module contains `kr8s`, a simple, extensible Python client library for Kubernetes.

   At the top level, `kr8s` provides a synchronous API that wraps the asynchronous API provided by `kr8s.asyncio`.
   Both APIs are functionally identical with the same objects, method signatures and return values.



Submodules
----------

.. toctree::
   :maxdepth: 1

   /autoapi/kr8s/asyncio/index
   /autoapi/kr8s/objects/index
   /autoapi/kr8s/portforward/index


Attributes
----------

.. autoapisummary::

   kr8s.ALL
   kr8s.KUBERNETES_MAXIMUM_SUPPORTED_VERSION
   kr8s.KUBERNETES_MINIMUM_SUPPORTED_VERSION
   kr8s.version
   kr8s.watch
   kr8s.api_resources


Exceptions
----------

.. autoapisummary::

   kr8s.APITimeoutError
   kr8s.ConnectionClosedError
   kr8s.ExecError
   kr8s.NotFoundError
   kr8s.ServerError


Classes
-------

.. autoapisummary::

   kr8s.Api


Functions
---------

.. autoapisummary::

   kr8s.get
   kr8s.api
   kr8s.whoami
   kr8s.create


Package Contents
----------------

.. py:data:: ALL
   :value: 'all'


.. py:data:: KUBERNETES_MAXIMUM_SUPPORTED_VERSION

.. py:data:: KUBERNETES_MINIMUM_SUPPORTED_VERSION

.. py:exception:: APITimeoutError

   Bases: :py:obj:`Exception`


   A timeout has occurred while waiting for a response from the Kubernetes API server.


.. py:exception:: ConnectionClosedError

   Bases: :py:obj:`Exception`


   A connection has been closed.


.. py:exception:: ExecError

   Bases: :py:obj:`Exception`


   Internal error in the exec protocol.


.. py:exception:: NotFoundError

   Bases: :py:obj:`Exception`


   Unable to find the requested resource.


.. py:exception:: ServerError(message: str, status: Optional[str] = None, response: Optional[httpx.Response] = None)

   Bases: :py:obj:`Exception`


   Error from the Kubernetes API server.

   Attributes:
       status: The Status object from the Kubernetes API server
       response: The httpx response object


   .. py:attribute:: status
      :value: None



   .. py:attribute:: response
      :value: None



.. py:class:: Api(**kwargs)

   Bases: :py:obj:`_api.Api`


   A kr8s object for interacting with the Kubernetes API.

   .. warning::
       This class is not intended to be instantiated directly. Instead, use the
       :func:`kr8s.api` function to get a singleton instance of the API.

       See https://docs.kr8s.org/en/stable/client.html#client-caching.



   .. py:method:: version() -> dict

      Get the Kubernetes version information from the API.

      Returns:
          The Kubernetes version information.




   .. py:method:: reauthenticate()

      Reauthenticate the API.



   .. py:method:: whoami()

      Retrieve the subject that's currently authenticated.

      Inspired by `kubectl whoami`.

      Returns:
          str: The subject that's currently authenticated.



   .. py:method:: lookup_kind(kind) -> tuple[str, str, bool]

      Lookup a Kubernetes resource kind.

      Check whether a resource kind exists on the remote server.

      Args:
          kind: The kind of resource to lookup.

      Returns:
          The kind of resource, the plural form and whether the resource is namespaced

      Raises:
          ValueError: If the kind is not found.



   .. py:method:: get(kind: str | type, *names: str, namespace: str | None = None, label_selector: str | dict | None = None, field_selector: str | dict | None = None, as_object: type[_objects.APIObject] | None = None, allow_unknown_type: bool = True, raw: bool = False, **kwargs) -> collections.abc.Generator[objects.APIObject | dict]

      Get Kubernetes resources.

      Args:
          kind: The kind of resource to get.
          *names: The names of specific resources to get.
          namespace: The namespace to get the resource from.
          label_selector: The label selector to filter the resources by.
          field_selector: The field selector to filter the resources by.
          as_object: The object to return the resources as.
          allow_unknown_type: Automatically create a class for the resource if none exists, default True.
          raw: If True, return raw dictionaries instead of APIObject instances, default False.
          **kwargs: Additional keyword arguments to pass to the API call.

      Returns:
          The resources.



   .. py:method:: watch(kind: str, namespace: str | None = None, label_selector: str | dict | None = None, field_selector: str | dict | None = None, since: str | None = None) -> collections.abc.Generator[tuple[str, objects.APIObject]]

      Watch a Kubernetes resource.



   .. py:method:: api_resources() -> list[dict]

      Get the Kubernetes API resources.



   .. py:method:: api_versions() -> collections.abc.Generator[str]

      Get the Kubernetes API versions.



   .. py:method:: create(resources: list[objects.APIObject])


   .. py:attribute:: auth


   .. py:property:: timeout


   .. py:method:: call_api(method: str = 'GET', version: str = 'v1', base: str = '', namespace: str | None = None, url: str = '', raise_for_status: bool = True, stream: bool = False, **kwargs) -> collections.abc.AsyncGenerator[httpx.Response]

      Make a Kubernetes API request.



   .. py:method:: open_websocket(version: str = 'v1', base: str = '', namespace: str | None = None, url: str = '', **kwargs) -> collections.abc.AsyncGenerator[httpx_ws.AsyncWebSocketSession]

      Open a websocket connection to a Kubernetes API endpoint.



   .. py:method:: async_version() -> dict
      :async:



   .. py:method:: async_reauthenticate() -> None
      :async:



   .. py:method:: async_whoami()
      :async:



   .. py:method:: async_lookup_kind(kind, skip_cache: bool = False) -> tuple[str, str, bool]
      :async:


      Lookup a Kubernetes resource kind.



   .. py:method:: async_get_kind(kind: str | type[kr8s._objects.APIObject], namespace: str | None = None, label_selector: str | dict | None = None, field_selector: str | dict | None = None, params: dict | None = None, watch: bool = False, allow_unknown_type: bool = True, **kwargs) -> collections.abc.AsyncGenerator[tuple[type[kr8s._objects.APIObject], httpx.Response]]
      :async:


      Get a Kubernetes resource.



   .. py:method:: async_get(kind: str | type, *names: str, namespace: str | None = None, label_selector: str | dict | None = None, field_selector: str | dict | None = None, as_object: type[kr8s._objects.APIObject] | None = None, allow_unknown_type: bool = True, raw: bool = False, **kwargs) -> collections.abc.AsyncGenerator[kr8s._objects.APIObject | dict]
      :async:



   .. py:method:: async_watch(kind: str, namespace: str | None = None, label_selector: str | dict | None = None, field_selector: str | dict | None = None, since: str | None = None, allow_unknown_type: bool = True) -> collections.abc.AsyncGenerator[tuple[str, kr8s._objects.APIObject]]
      :async:


      Watch a Kubernetes resource.



   .. py:method:: async_api_resources() -> list[dict]
      :async:



   .. py:method:: async_api_resources_uncached() -> list[dict]
      :async:


      Get the Kubernetes API resources.



   .. py:method:: async_api_versions() -> collections.abc.AsyncGenerator[str]
      :async:



   .. py:method:: async_create(resources: list[kr8s._objects.APIObject])
      :async:



   .. py:property:: namespace
      :type: str


      Get the default namespace.



.. py:function:: get(kind: str, *names: str, namespace: str | None = None, label_selector: str | dict | None = None, field_selector: str | dict | None = None, as_object: type | None = None, allow_unknown_type: bool = True, raw: bool = False, api=None, **kwargs)

   Get a resource by name.

   Args:
       kind: The kind of resource to get
       *names: The names of the resources to get
       namespace: The namespace to get the resource from
       label_selector: The label selector to filter the resources by
       field_selector: The field selector to filter the resources by
       as_object: The object to populate with the resource data
       allow_unknown_type: Whether to allow unknown types
       raw: If True, return raw dictionaries instead of APIObject instances, default False
       api: The api to use to get the resource
       **kwargs: Additional arguments to pass to the API

   Returns:
       The populated object (or dict if raw=True)

   Raises:
       ValueError: If the resource is not found

   Examples:
       >>> import kr8s
       >>> # All of these are equivalent
       >>> ings = kr8s.get("ing")                           # Short name
       >>> ings = kr8s.get("ingress")                       # Singular
       >>> ings = kr8s.get("ingresses")                     # Plural
       >>> ings = kr8s.get("Ingress")                       # Title
       >>> ings = kr8s.get("ingress.networking.k8s.io")     # Full group name
       >>> ings = kr8s.get("ingress.v1.networking.k8s.io")  # Full with explicit version
       >>> ings = kr8s.get("ingress.networking.k8s.io/v1")  # Full with explicit version alt.
       >>>
       >>> # Get raw dictionaries for better performance
       >>> pods = list(kr8s.get("pods", raw=True))
       >>> print(pods[0]["metadata"]["name"])


.. py:function:: api(url: str | None = None, kubeconfig: str | None = None, serviceaccount: str | None = None, namespace: str | None = None, context: str | None = None) -> Api

   Create a :class:`kr8s.Api` object for interacting with the Kubernetes API.

   If a kr8s object already exists with the same arguments in this thread, it will be returned.

   Args:
       url: The URL of the Kubernetes API server
       kubeconfig: The path to a kubeconfig file to use
       serviceaccount: The path of a service account to use
       namespace: The namespace to use
       context: The context to use

   Returns:
       The API object

   Examples:
       >>> import kr8s
       >>> api = kr8s.api()  # Uses the default kubeconfig
       >>> print(api.version())  # Get the Kubernetes version


.. py:function:: whoami()

   Get the current user's identity.

   Returns:
       The user's identity

   Examples:
       >>> import kr8s
       >>> print(kr8s.whoami())


.. py:function:: create(resources: list[type[_objects.APIObject]], api=None)

   Creates resources in the Kubernetes cluster.


.. py:data:: version

.. py:data:: watch

.. py:data:: api_resources