kr8s

Warnings

• gotcha Unlike some auto-generated Kubernetes client libraries, kr8s is designed with a `kubectl`-like API. It simplifies interactions and makes assumptions about authentication and API versions from your environment/kubeconfig. This is a design choice to reduce boilerplate but may differ from expectations if coming from more verbose, low-level clients.
  Fix: Familiarize yourself with the kr8s design philosophy as outlined in its documentation. Explicit API client creation (e.g., `api = kr8s.api()`) allows for more control if needed.
• gotcha kr8s's object model is an extensible, flexible, minimal set, not a 1:1 reflection of the full Kubernetes OpenAPI schema. While it covers common resources, complex or custom resource definitions might require direct API calls or subclassing `kr8s.objects.APIObject` for full fidelity. For strict typing and schema validation, consider integrating with libraries like `Lightkube`.
  Fix: Refer to `kr8s` documentation on `APIObject` for extending its resource model. For strict validation, consider using Lightkube models or direct dictionary manipulation for spec definitions.
• gotcha kr8s provides both synchronous (`import kr8s`) and asynchronous (`import kr8s.asyncio`) APIs. The synchronous API is a wrapper around the asynchronous one. Mixing them incorrectly or using the synchronous API within an existing `asyncio` event loop without proper handling can lead to blocking or unexpected behavior.
  Fix: Choose either the synchronous or asynchronous API for a given code path. When using `kr8s` in an `asyncio` application, prefer `kr8s.asyncio` for non-blocking operations. When integrating sync `kr8s` with async code, consider `anyio.to_thread.run_sync` if you need to call blocking kr8s functions.
• breaking kr8s aims to support actively maintained Kubernetes versions. As Kubernetes versions reach their end-of-life (EOL) from the community and cloud providers, kr8s may drop official support and testing for these versions. This could lead to compatibility issues for users on older, unsupported Kubernetes clusters.
  Fix: Ensure your Kubernetes cluster is running a currently supported version. Regularly check the Kubernetes project's and your cloud provider's EOL policies, and the `kr8s` version support documentation.
• deprecated In version 0.20.15, a change was implemented to "Use `content` arg instead of `data` when calling api". This implies that using the `data` argument in direct API calls might be deprecated or will be removed in future versions.
  Fix: Update any custom direct API calls from `api(..., data=...)` to `api(..., content=...)` to ensure future compatibility. Consult the changelog for specific API methods affected.

Install

• pip install kr8s Install with pip

Imports

• kr8s

  import kr8s

  Primary synchronous API entry point.
• Pod

  from kr8s.objects import Pod

  Import specific Kubernetes resource objects.
• asyncio

  import kr8s.asyncio

  Entry point for the asynchronous API.

Quickstart

This quickstart demonstrates basic synchronous interaction with a Kubernetes cluster using `kr8s`. It retrieves a list of nodes, creates a simple Nginx pod, waits for it to become ready, and then deletes it. kr8s automatically uses your existing kubeconfig for authentication, similar to `kubectl`.

import kr8s

# Connect to the Kubernetes API (uses kubeconfig by default)
# Equivalent to 'kubectl get nodes'

try:
    # Synchronous API
    nodes = kr8s.get("nodes")
    print("Synchronous API - Nodes found:")
    for node in nodes:
        print(f"- {node.name}")

    # Example using a specific object type and creating one
    from kr8s.objects import Pod

    # Define a simple Pod specification as a Python dictionary
    pod_manifest = {
        "apiVersion": "v1",
        "kind": "Pod",
        "metadata": {
            "name": "my-test-pod",
            "labels": {"app": "my-app"}
        },
        "spec": {
            "containers": [
                {"name": "nginx", "image": "nginx:latest"}
            ]
        }
    }

    print(f"\nAttempting to create pod: {pod_manifest['metadata']['name']}")
    test_pod = Pod(pod_manifest)
    test_pod.create()
    print(f"Pod '{test_pod.name}' created in namespace '{test_pod.namespace}'.")

    # Wait for the pod to be ready (optional)
    print("Waiting for pod to be ready...")
    test_pod.wait(conditions=["Ready"], timeout=60)
    print("Pod is ready!")

    # Clean up the pod
    print(f"\nDeleting pod: {test_pod.name}")
    test_pod.delete(cascade=True)
    print("Pod deleted.")

except Exception as e:
    print(f"An error occurred: {e}")
    print("Ensure you have a running Kubernetes cluster and valid kubeconfig configured (e.g., `minikube start`).")