### Complete example: Patching StatefulSet resources and comparing

Source: https://github.com/gtsystem/lightkube/blob/master/docs/utils.md

This comprehensive example demonstrates patching a StatefulSet's resource limits and requests, then verifying the changes by comparing the active pod's resources against the desired state using `parse_quantity` and `equals_canonically`. It covers the entire workflow from defining resource requirements to applying a patch and asserting equality.

```python
>>> from lightkube import Client
>>> from lightkube.models.apps_v1 import StatefulSetSpec
>>> from lightkube.models.core_v1 import (Container, PodSpec, PodTemplateSpec, ResourceRequirements)
>>> from lightkube.resources.apps_v1 import StatefulSet
>>> from lightkube.resources.core_v1 import Pod
>>> from lightkube.types import PatchType
>>>
>>> resource_reqs = ResourceRequirements(
...     limits={"cpu": "0.8", "memory": "0.9Gi"},
...     requests={"cpu": "0.4", "memory": "0.5Gi"},
... )
>>>
>>> client = Client()
>>> statefulset = client.get(StatefulSet, name="prom")
>>>
>>> delta = StatefulSet(
...     spec=StatefulSetSpec(
...         selector=statefulset.spec.selector,
...         serviceName=statefulset.spec.serviceName,
...         template=PodTemplateSpec(
...             spec=PodSpec(
...                 containers=[Container(name="prometheus", resources=resource_reqs)]
...             )
...         )
...     )
... )
>>>
>>> client.patch(StatefulSet, "prom", delta, patch_type=PatchType.APPLY, field_manager="just me")
>>> client.get(StatefulSet, name="prom").spec.template.spec.containers[1].resources
ResourceRequirements(limits={'cpu': '800m', 'memory': '966367641600m'}, requests={'cpu': '400m', 'memory': '512Mi'})
>>>
>>> pod = client.get(Pod, name="prom-0")
>>> pod.spec.containers[1].resources
ResourceRequirements(limits={'cpu': '800m', 'memory': '966367641600m'}, requests={'cpu': '400m', 'memory': '512Mi'})
>>>
>>> from lightkube.utils.quantity import parse_quantity
>>> parse_quantity(pod.spec.containers[1].resources.requests["memory"])
Decimal('536870912.000')
>>> parse_quantity(pod.spec.containers[1].resources.requests["memory"]) == parse_quantity(resource_reqs.requests["memory"])
True
>>>
>>> from lightkube.utils.quantity import equals_canonically
>>> equals_canonically(pod.spec.containers[1].resources.limits, resource_reqs.limits)
True
>>> equals_canonically(pod.spec.containers[1].resources, resource_reqs)
True
```

--------------------------------

### Get a Deployment Resource using lightkube Client

Source: https://github.com/gtsystem/lightkube/blob/master/docs/resources-and-models.md

Demonstrates how to retrieve a specific Kubernetes Deployment resource by name using the lightkube client. Ensure the `lightkube` library is installed and configured to connect to your Kubernetes cluster.

```python
>>> from lightkube import Client
>>> from lightkube.resources.apps_v1 import Deployment

>>> client = Client()
>>> dep = client.get(Deployment, name="my-deo")
>>> type(dep)
<class 'lightkube.resources.apps_v1.Deployment'>
```

--------------------------------

### Install lightkube

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Install the lightkube module using pip. Requires Python 3.8 or higher.

```sh
pip install lightkube
```

--------------------------------

### Install lightkube using uv

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Install the lightkube package using uv. Requires Python 3.8 or higher.

```sh
uv add lightkube
```

--------------------------------

### Scale Deployment (Async)

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Asynchronously scales a Kubernetes Deployment. Requires importing AsyncClient and related models. The example is wrapped in an async function.

```python
from lightkube import AsyncClient
from lightkube.resources.apps_v1 import Deployment
from lightkube.models.meta_v1 import ObjectMeta
from lightkube.models.autoscaling_v1 import ScaleSpec

async def example():
    client = AsyncClient()
    obj = Deployment.Scale(
        metadata=ObjectMeta(name='metrics-server', namespace='kube-system'),
        spec=ScaleSpec(replicas=1)
    )
    await client.replace(obj, 'metrics-server', namespace='kube-system')
```

--------------------------------

### Get a Pod by Name

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Retrieves a specific pod from the 'default' namespace. Requires the 'lightkube' library.

```python
from lightkube import Client
from lightkube.resources.core_v1 import Pod

client = Client()
pod = client.get(Pod, name="my-pod", namespace="default")
print(pod.namespace.uid)
```

--------------------------------

### Implement List-Watch Pattern with lightkube

Source: https://github.com/gtsystem/lightkube/blob/master/docs/list-watch.md

Use this pattern to efficiently detect changes in Kubernetes resources. It requires iterating through a list to obtain the `resourceVersion` before starting the watch. Accessing `resourceVersion` before iteration will raise a `lightkube.NotReadyError`.

```python
seen_pods = {}
async for pod in (podlist := client.list(Pod)):
    seen_pods[pod.metadata.name] = pod
async for event, pod in client.watch(Pod, resource_version=podlist.resourceVersion):
    match event:
        case "ADDED" | "MODIFIED":
            seen_pods[pod.metadata.name] = pod
        case "DELETED":
            del seen_pods[pod.metadata.name]
```

--------------------------------

### Field Selector: Match by Status Phase

Source: https://github.com/gtsystem/lightkube/blob/master/docs/selectors.md

Select resources based on their status phase, for example, 'Pending'.

```python
fields={"status.phase": "Pending"}
```

--------------------------------

### Server-Side Apply ConfigMap (Async)

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Asynchronously applies a ConfigMap using server-side apply. Requires specifying a field manager and providing apiVersion/kind in the object definition. The example is wrapped in an async function.

```python
from lightkube import AsyncClient
from lightkube.resources.core_v1 import ConfigMap
from lightkube.models.meta_v1 import ObjectMeta

async def example():
    client = AsyncClient(field_manager="my-manager")
    config = ConfigMap(
        apiVersion='v1', kind='ConfigMap',
        metadata=ObjectMeta(name='my-config', namespace='default'),
        data={'key1': 'value1', 'key2': 'value2'}
    )

    res = await client.apply(config)
    print(res.data)
    # prints {'key1': 'value1', 'key2': 'value2'}

    del config.data['key1']
    config.data['key3'] = 'value3'

    res = await client.apply(config)
    print(res.data)
    # prints {'key2': 'value2', 'key3': 'value3'}
```

--------------------------------

### Server-Side Apply for ConfigMap

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Applies a ConfigMap using server-side apply, which requires a field manager. The 'apiVersion' and 'kind' must be specified in the object. This example shows initial creation and then modification.

```python
from lightkube.resources.core_v1 import ConfigMap
from lightkube.models.meta_v1 import ObjectMeta

client = Client(field_manager="my-manager")
config = ConfigMap(
    # note apiVersion and kind need to be specified for server-side apply
    apiVersion='v1', kind='ConfigMap',
    metadata=ObjectMeta(name='my-config', namespace='default'),
    data={'key1': 'value1', 'key2': 'value2'}
)

res = client.apply(config)
print(res.data)
# prints {'key1': 'value1', 'key2': 'value2'}

del config.data['key1']
config.data['key3'] = 'value3'

res = client.apply(config)
print(res.data)
# prints {'key2': 'value2', 'key3': 'value3'}
```

--------------------------------

### Update Deployment Status (Async)

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Asynchronously updates the status of a Kubernetes Deployment. Requires importing AsyncClient and DeploymentStatus. The example is wrapped in an async function.

```python
from lightkube import AsyncClient
from lightkube.resources.apps_v1 import Deployment
from lightkube.models.apps_v1 import DeploymentStatus

async def example():
    client = AsyncClient()
    obj = Deployment.Status(
        status=DeploymentStatus(observedGeneration=99)
    )
    await client.apply(obj, name='metrics-server', namespace='kube-system')
```

--------------------------------

### Getting Generic Resource

Source: https://github.com/gtsystem/lightkube/blob/master/docs/generic-resources.md

Details the `get_generic_resource` function for retrieving a specific generic resource.

```APIDOC
## get_generic_resource

### Description
Retrieves a specific generic resource from the cluster.

### Method
`get_generic_resource(client: Client, group: str, version: str, kind: str, plural: str, name: str, namespace: Optional[str] = None)`

### Parameters
- **client** (Client) - The lightkube client instance.
- **group** (str) - The API group of the resource.
- **version** (str) - The API version of the resource.
- **kind** (str) - The Kind of the resource.
- **plural** (str) - The plural name of the resource.
- **name** (str) - The name of the resource to retrieve.
- **namespace** (Optional[str]) - The namespace of the resource (if applicable).
```

--------------------------------

### Create and Get Namespaced Generic Resource

Source: https://github.com/gtsystem/lightkube/blob/master/docs/generic-resources.md

Define a generic resource type for a namespaced resource like a Job and then retrieve an instance of it from the cluster.

```python
from lightkube import Client
from lightkube.generic_resource import create_namespaced_resource

Job = create_namespaced_resource('stable.example.com', 'v1', 'Job', 'jobs')

client = Client()
job = client.get(Job, name="job1", namespace="my-namespace")
```

--------------------------------

### Get a Pod asynchronously

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Retrieve a specific Pod resource by name and namespace using the asynchronous client. Accesses the Pod's namespace UID.

```python
from lightkube import AsyncClient
from lightkube.resources.core_v1 import Pod

async def example():
    client = AsyncClient()
    pod = await client.get(Pod, name="my-pod", namespace="default")
    print(pod.namespace.uid)
```

--------------------------------

### Update Deployment Status

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Updates the status of a Kubernetes Deployment, for example, the observed generation. Requires 'lightkube', 'Deployment.Status', and 'DeploymentStatus'.

```python
from lightkube import Client
from lightkube.resources.apps_v1 import Deployment
from lightkube.models.apps_v1 import DeploymentStatus

client = Client()
obj = Deployment.Status(
    status=DeploymentStatus(observedGeneration=99)
)
client.apply(obj, name='metrics-server', namespace='kube-system')
```

--------------------------------

### Get Generic Resource Subresource

Source: https://github.com/gtsystem/lightkube/blob/master/docs/generic-resources.md

Retrieve a subresource, such as 'Status', for a generic resource. Note that not all resources support all subresources like 'Scale'.

```python
job = client.get(Job.Status, name="job1", namespace="my-namespace")
```

--------------------------------

### Auto-Detect Configuration from Environment

Source: https://github.com/gtsystem/lightkube/blob/master/docs/configuration.md

Initialize a client without providing explicit configuration; lightkube will auto-detect settings from the environment.

```python
import lightkube

client = lightkube.Client() # no configuration provided
```

```python
from lightkube import KubeConfig, Client

config = KubeConfig.from_env()
client = Client(config=config)
```

--------------------------------

### Create a ConfigMap

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Creates a new ConfigMap resource in the 'default' namespace with specified data. Requires 'lightkube' and 'ObjectMeta' from 'lightkube.models.meta_v1'.

```python
from lightkube import Client
from lightkube.resources.core_v1 import ConfigMap
from lightkube.models.meta_v1 import ObjectMeta

client = Client()
config = ConfigMap(
    metadata=ObjectMeta(name='my-config', namespace='default'),
    data={'key1': 'value1', 'key2': 'value2'}
)

client.create(config)
```

--------------------------------

### Load KubeConfig and Select Context

Source: https://github.com/gtsystem/lightkube/blob/master/docs/configuration.md

Load a Kubernetes configuration from a file and explicitly select a context to use.

```python
from lightkube import KubeConfig, Client

config = KubeConfig.from_file("path/to/my/config")
client = Client(config=config.get())
```

```python
# use the context named my-context
client = Client(config=config.get(context_name='my-context'))
```

--------------------------------

### Watch Deployments in a Namespace

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Streams events for Deployment resources in the 'default' namespace. Requires 'lightkube' and 'Deployment'.

```python
from lightkube import Client
from lightkube.resources.apps_v1 import Deployment

client = Client()
for op, dep in client.watch(Deployment, namespace="default"):
    print(f"{dep.namespace.name} {dep.spec.replicas}")
```

--------------------------------

### Creating Generic Resources Manually

Source: https://github.com/gtsystem/lightkube/blob/master/docs/generic-resources.md

Illustrates how to manually create a generic resource object and then create it in the cluster.

```APIDOC
## Creating Generic Resources Manually

### Description
You can manually instantiate a generic resource object and then use the client to create it in the cluster.

### Example
```python
# Assuming Job is a previously created generic resource type
job = Job(metadata={"name": "job2", "namespace": "my-namespace"}, spec=...)
client.create(job)
```

### Note
Since generic resources are schemaless, ensure you provide valid attributes to avoid server errors.
```

--------------------------------

### Instantiate an ObjectMeta Model

Source: https://github.com/gtsystem/lightkube/blob/master/docs/resources-and-models.md

Shows how to create an instance of the `ObjectMeta` model from `lightkube.models.meta_v1`. This model is used to represent metadata for Kubernetes objects, such as name and namespace.

```python
>>> from lightkube.models.meta_v1 import ObjectMeta
>>> ObjectMeta(name='test', namespace='default')
ObjectMeta(annotations=None, clusterName=None, creationTimestamp=None, deletionGracePeriodSeconds=None, deletionTimestamp=None, finalizers=None, generateName=None, generation=None, initializers=None, labels=None, managedFields=None, name='test', namespace='default', ownerReferences=None, resourceVersion=None, selfLink=None, uid=None)
```

--------------------------------

### Creating Namespaced and Global Resources

Source: https://github.com/gtsystem/lightkube/blob/master/docs/generic-resources.md

Demonstrates how to create namespaced and global generic resources using `create_namespaced_resource` and `create_global_resource`.

```APIDOC
## create_namespaced_resource

### Description
Creates a namespaced generic resource.

### Method
`create_namespaced_resource(group: str, version: str, kind: str, plural: str)`

## create_global_resource

### Description
Creates a global generic resource.

### Method
`create_global_resource(group: str, version: str, kind: str, plural: str)`
```

--------------------------------

### Load KubeConfig from File

Source: https://github.com/gtsystem/lightkube/blob/master/docs/configuration.md

Load a specific Kubernetes configuration from a YAML file. The client will use the current context by default.

```python
from lightkube import KubeConfig, Client

config = KubeConfig.from_file("path/to/my/config")
client = Client(config=config)
```

--------------------------------

### Watch Deployments (Async)

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Use the async client to asynchronously watch for changes in Deployments within a specified namespace.

```python
from lightkube import AsyncClient
from lightkube.resources.apps_v1 import Deployment

async def example():
    client = AsyncClient()
    async for op, dep in client.watch(Deployment, namespace="default"):
        print(f"{dep.namespace.name} {dep.spec.replicas}")
```

--------------------------------

### Execute Command in Pod (Sync)

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Use the sync client to execute commands inside a pod, capturing stdout and handling errors. Supports sending data to stdin.

```python
from lightkube import Client

client = Client()

# Capture stdout or raise ApiError if error code is != 0
res = client.exec('my-pod', namespace='default', container='main', command=['ls', '-l', '/'], 
    stdout=True, raise_on_error=True)
print(res.stdout)

# Send data to stdin and capture output
res = client.exec('my-pod', namespace='default', container='main', command=['cat'], 
    stdin='hello\n', stdout=True)
print(res.stdout)
print(res.exit_code)
```

--------------------------------

### List All Nodes

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Lists all nodes in the cluster. Requires the 'lightkube' library.

```python
from lightkube import Client
from lightkube.resources.core_v1 import Node

client = Client()
for node in client.list(Node):
    print(node.metadata.name)
```

--------------------------------

### Execute Command in Pod (stdin/stdout)

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Executes a command inside a specified pod and container, sending data to stdin and capturing stdout and the exit code. Requires 'lightkube'.

```python
from lightkube import Client

client = Client()

# Send data to stdin and capture output
res = client.exec('my-pod', namespace='default', container='main', command=['cat'], 
    stdin='hello\n', stdout=True)
print(res.stdout)
print(res.exit_code)
```

--------------------------------

### Create Custom Resource Classes

Source: https://github.com/gtsystem/lightkube/blob/master/docs/custom-resources.md

Create Resource subclasses that integrate your custom models with lightkube's client. This involves defining ApiInfo and registering the resource. Include a Status subclass if your resource has a status subresource.

```python
from typing import ClassVar

from lightkube.codecs import resource_registry
from lightkube.core import resource as res

from ..models import dog as m_dog


# Only needed if your custom resource has a status subresource
class DogStatus(res.NamespacedSubResource, m_dog.Dog):
    _api_info = res.ApiInfo(
        resource=res.ResourceDef('stable.example.com', 'v1', 'Dog'),
        parent=res.ResourceDef('stable.example.com', 'v1', 'Dog'),
        plural='dogs',
        verbs=['get', 'patch', 'put'],
        action='status',
    )


@resource_registry.register
class Dog(res.NamespacedResourceG, m_dog.Dog):
    _api_info = res.ApiInfo(
        resource=res.ResourceDef('stable.example.com', 'v1', 'Dog'),
        plural='dogs',
        verbs=[
            'delete', 'deletecollection', 'get', 'global_list', 'global_watch', 
            'list', 'patch', 'post', 'put', 'watch'
        ],
    )

    # Only needed if your custom resource has a status subresource
    Status: ClassVar = DogStatus
```

--------------------------------

### Create a ConfigMap asynchronously

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Create a ConfigMap resource with specified metadata and data using the asynchronous client.

```python
from lightkube import AsyncClient
from lightkube.resources.core_v1 import ConfigMap
from lightkube.models.meta_v1 import ObjectMeta

async def example():
    client = AsyncClient()
    config = ConfigMap(
        metadata=ObjectMeta(name='my-config', namespace='default'),
        data={'key1': 'value1', 'key2': 'value2'}
    )
    await client.create(config)
```

--------------------------------

### Create Resources from YAML File

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Reads and creates Kubernetes resources defined in a YAML file. Requires 'lightkube' and 'codecs'.

```python
from lightkube import Client, codecs

client = Client()
with open('deployment.yaml') as f:
    for obj in codecs.load_all_yaml(f):
        client.create(obj)
```

--------------------------------

### Convert ObjectMeta to Dictionary

Source: https://github.com/gtsystem/lightkube/blob/master/docs/codecs.md

Demonstrates creating an ObjectMeta model from a dictionary and converting it back to a dictionary.

```python
from lightkube.models.meta_v1 import ObjectMeta
meta = ObjectMeta.from_dict({'name': 'my-name', 'labels': {'key': 'value'}})
```

```python
meta_dict = meta.to_dict()
```

--------------------------------

### Field Selector: Match by Name

Source: https://github.com/gtsystem/lightkube/blob/master/docs/selectors.md

Use this to select resources based on their metadata name.

```python
fields={"metadata.name": "myobj"}
```

--------------------------------

### Create Generic Resource Manually

Source: https://github.com/gtsystem/lightkube/blob/master/docs/generic-resources.md

Manually construct a generic resource object and create it in the cluster. Ensure all required fields like metadata and spec are provided.

```python
job = Job(metadata={"name": "job2", "namespace": "my-namespace"}, spec=...)
client.create(job)
```

--------------------------------

### Stream Pod Logs

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Follows and prints logs from a specific pod in real-time. Requires 'lightkube'.

```python
from lightkube import Client

client = Client()
for line in client.log('my-pod', follow=True):
    print(line)
```

--------------------------------

### Execute Command in Pod (Async)

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Use the async client to asynchronously execute commands inside a pod, capturing stdout and handling errors. Supports sending data to stdin.

```python
from lightkube import AsyncClient

async def example():
    client = AsyncClient()

    # List a directory
    res = await client.exec('my-pod', namespace='default', container='main', command=['ls', '-l', '/'], 
        stdout=True, raise_on_error=True)
    print(res.stdout)

    # Send data to stdin and capture output
    res = await client.exec('my-pod', namespace='default', container='main', command=['cat'], 
        stdin='hello\n', stdout=True)
    print(res.stdout)
    print(res.exit_code)
```

--------------------------------

### Handle Lightkube Configuration Errors

Source: https://github.com/gtsystem/lightkube/blob/master/docs/reference/exceptions.md

Catch ConfigError when initializing the Lightkube client if there are issues with the Kubernetes configuration.

```python
from lightkube import Client, ConfigError

try:
    client = Client()
except ConfigError as e:
    print(e)
```

--------------------------------

### Scale a Deployment

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Scales a Kubernetes Deployment to a specified number of replicas. Requires 'lightkube', 'Deployment.Scale', 'ObjectMeta', and 'ScaleSpec'.

```python
from lightkube import Client
from lightkube.resources.apps_v1 import Deployment
from lightkube.models.meta_v1 import ObjectMeta
from lightkube.models.autoscaling_v1 import ScaleSpec

client = Client()
obj = Deployment.Scale(
    metadata=ObjectMeta(name='metrics-server', namespace='kube-system'),
    spec=ScaleSpec(replicas=1)
)
client.replace(obj)
```

--------------------------------

### Create resources from YAML file asynchronously

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Load and create multiple Kubernetes resources defined in a YAML file using the asynchronous client.

```python
from lightkube import AsyncClient, codecs

async def example():
    client = AsyncClient()
    with open('deployment.yaml') as f:
        for obj in codecs.load_all_yaml(f):
            await client.create(obj)
```

--------------------------------

### Sorting Objects for Apply

Source: https://github.com/gtsystem/lightkube/blob/master/docs/codecs.md

Demonstrates how to use `sort_objects` to order Kubernetes resources from a manifest for batch apply operations, ensuring dependencies are met.

```APIDOC
## Sorting resource objects

Sometimes you have a manifest of resources where some depend on others.  For example, consider the following `yaml_with_dependencies.yaml` file:

```yaml
kind: ClusterRoleBinding
roleRef:
  kind: ClusterRole
  name: example-cluster-binding
subjects:
  - kind: ServiceAccount
    name: example-service-account
...
---
kind: ClusterRole
metadata:
  name: example-cluster-role
...
---
kind: ServiceAccount
metadata:
  name: example-service-account
```

where we have a `ClusterRoleBinding` that uses a `ClusterRole` and `ServiceAccount`. In cases like this, the order in which we `apply` these resources matters as the `ClusterRoleBinding` depends on the others. To sort these objects so that we do not encounter API errors when `apply`ing them, use `sort_objects(...)`.

::: lightkube.sort_objects
    options:
        heading_level: 3

Revisiting the example above, we can apply from `yaml_with_dependencies.yaml` by:

```python
from lightkube import Client, codecs, sort_objects

client = Client()
with open('yaml_with_dependencies.yaml') as f:
    objects = codecs.load_all_yaml(f)
    for obj in sort_objects(objects):
        client.create(obj)
```

`sort_objects` orders the objects in a way that is friendly to applying them as a batch, allowing us to loop through them as normal.
```

--------------------------------

### Load In-Cluster Configuration

Source: https://github.com/gtsystem/lightkube/blob/master/docs/configuration.md

Build a Kubernetes configuration from the service account data available within a pod. This configuration automatically supports token refresh.

```python
from lightkube import KubeConfig, Client

config = KubeConfig.from_service_account()
client = Client(config=config)
```

--------------------------------

### Execute Command in Pod (stdout)

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Executes a command inside a specified pod and container, capturing stdout. If the command returns a non-zero exit code, an ApiError is raised. Requires 'lightkube'.

```python
from lightkube import Client

client = Client()

# Capture stdout or raise ApiError if error code is != 0
res = client.exec('my-pod', namespace='default', container='main', command=['ls', '-l', '/'], 
    stdout=True, raise_on_error=True)
print(res.stdout)
```

--------------------------------

### Register Custom Resource with codecs.resource_registry

Source: https://github.com/gtsystem/lightkube/blob/master/docs/codecs.md

Register a custom resource with `codecs.resource_registry.register()` so it can be loaded by lightkube's YAML functions. This can be done directly or as a decorator.

```python
from lightkube import codecs

codecs.resource_registry.register(MyCustomResource)

with open('service.yaml') as f:
    # Now `MyCustomResource` can be loaded
    objs = codecs.load_all_yaml(f)
```

```python
from lightkube.core.resource import NamespacedResource
from lightkube.codecs import resource_registry

@resource_registry.register
class MyCustomResource(NamespacedResource):
    ...
```

--------------------------------

### Load Resource Object from Dictionary

Source: https://github.com/gtsystem/lightkube/blob/master/docs/codecs.md

Loads a Kubernetes resource object dynamically from a dictionary representation using codecs.from_dict. Only known resources can be loaded.

```python
from lightkube import codecs

obj = codecs.from_dict({
        'apiVersion': 'v1',
        'kind': 'ConfigMap',
        'metadata': {'name': 'config-name', 'labels': {'label1': 'value1'}},
        'data': {
            'file1.txt': 'some content here',
            'file2.txt': 'some other content'
        }
})
print(type(obj))
```

--------------------------------

### Accessing Subresources

Source: https://github.com/gtsystem/lightkube/blob/master/docs/generic-resources.md

Explains how to access subresources like 'Status' and 'Scale' for generic resources.

```APIDOC
## Accessing Subresources

### Description
Subresources such as `Status` and `Scale` can be accessed for generic resources.

### Example
```python
# Accessing the Status subresource
job = client.get(Job.Status, name="job1", namespace="my-namespace")
```

### Note
Support for the `Scale` subresource may vary depending on the specific resource.
```

--------------------------------

### Label Selector: Using Operator Functions

Source: https://github.com/gtsystem/lightkube/blob/master/docs/selectors.md

Import and use operator functions for more complex label matching logic.

```python
from lightkube import operators as op
```

--------------------------------

### Stream Pod Logs (Async)

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Use the async client to asynchronously stream logs from a specific pod.

```python
from lightkube import AsyncClient

async def example():
    client = AsyncClient()
    async for line in client.log('my-pod', follow=True):
        print(line)
```

--------------------------------

### Load All YAML Resources from File

Source: https://github.com/gtsystem/lightkube/blob/master/docs/codecs.md

Loads all Kubernetes resource objects defined in a YAML file. Supports creating resources for CRDs if specified.

```python
from lightkube import Client, codecs

client = Client()
with open('deployment.yaml') as f:
    for obj in codecs.load_all_yaml(f):
        client.create(obj)
```

```python
from lightkube import Client, codecs

client = Client()
with open('file-with-crd-and-instance.yaml') as f:
    for obj in codecs.load_all_yaml(f, create_resources_for_crds=True):
        client.create(obj)
```

--------------------------------

### List Nodes asynchronously

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

List all Node resources in the cluster using the asynchronous client. Prints the name of each node.

```python
from lightkube import AsyncClient
from lightkube.resources.core_v1 import Node

async def example():
    client = AsyncClient()
    async for node in client.list(Node):
        print(node.metadata.name)
```

--------------------------------

### Load All YAML and Then Generic Resources

Source: https://github.com/gtsystem/lightkube/blob/master/docs/generic-resources.md

Load resources from a YAML file and then load all Custom Resource Definitions (CRDs) from the cluster as generic resources. This allows subsequent YAML loading to recognize custom resources.

```yaml
apiVersion: "stable.example.com/v1"
kind: CronTab
metadata:
  name: my-new-cron-object
spec:
  cronSpec: "* * * * */5"
  image: my-awesome-cron-image
```

```python
from pathlib import Path

from lightkube import Client
from lightkube.codecs import load_all_yaml
from lightkube.generic_resource import load_in_cluster_generic_resources

# This fails with error message:
# lightkube.core.exceptions.LoadResourceError: No module named 'lightkube.resources.stable_example_com_v1'. If using a CRD, ensure you define a generic resource.
resources = load_all_yaml(Path("crontab.yaml").read_text())

client = Client()
load_in_cluster_generic_resources(client)

# Now we can load_all_yaml (and use those loaded resources, for example to create them in cluster)
resources = load_all_yaml(Path("crontab.yaml").read_text())
```

--------------------------------

### Resource Registry Registration

Source: https://github.com/gtsystem/lightkube/blob/master/docs/codecs.md

Explains how to use the `resource_registry` to register custom resources, enabling them to be loaded by the codec functions.

```APIDOC
## Resource Registry

::: lightkube.codecs.resource_registry
    options:
        heading_level: 3

The singleton `resource_registry` allows to register a custom resource, so that it can be used by the load functions on this module:

```python
from lightkube import codecs

codecs.resource_registry.register(MyCustomResource)

with open('service.yaml') as f:
    # Now `MyCustomResource` can be loaded
    objs = codecs.load_all_yaml(f)
```

`register` can also be used as a decorator:
```python
from lightkube.core.resource import NamespacedResource
from lightkube.codecs import resource_registry

@resource_registry.register
class MyCustomResource(NamespacedResource):
    ...
```
```

--------------------------------

### Dump Resources to YAML File

Source: https://github.com/gtsystem/lightkube/blob/master/docs/codecs.md

Dumps a list of Kubernetes resource objects into a YAML file. Ensure the file is opened in write mode.

```python
from lightkube.resources.core_v1 import ConfigMap
from lightkube.models.meta_v1 import ObjectMeta
from lightkube import codecs

cm = ConfigMap(
    apiVersion='v1', kind='ConfigMap',
    metadata=ObjectMeta(name='xyz', labels={'x': 'y'})
)
with open('deployment-out.yaml', 'w') as fw:
    codecs.dump_all_yaml([cm], fw)
```

--------------------------------

### Accessing Generic Resource Data

Source: https://github.com/gtsystem/lightkube/blob/master/docs/generic-resources.md

Shows how to access data within a generic resource, both as a dictionary and using attribute notation.

```APIDOC
## Accessing Generic Resource Data

### Description
Generic resources are subclasses of `dict`, allowing access to their content using dictionary key access or attribute notation for common fields like `apiVersion`, `metadata`, `kind`, and `status`.

### Example
```python
# Accessing data as a dictionary
print(job["path"]["to"]["something"])

# Accessing common attributes
print(job.kind)
print(job.metadata)

# Accessing nested metadata fields
print(job.metadata.name)
```

### Note
Metadata is decoded using the `models.meta_v1.ObjectMeta` model.
```

--------------------------------

### Connect to Kubernetes API via Proxy

Source: https://github.com/gtsystem/lightkube/blob/master/docs/configuration.md

Build a configuration to connect to a non-protected Kubernetes API, useful for tunneling API calls through kubectl proxy.

```bash
kubectl proxy --port=8080
```

```python
from lightkube import KubeConfig, Client

config = KubeConfig.from_server("http://localhost:8080")
client = Client(config=config)
```

--------------------------------

### Loading In-Cluster Generic Resources

Source: https://github.com/gtsystem/lightkube/blob/master/docs/generic-resources.md

Demonstrates using `load_in_cluster_generic_resources` to load all CRDs in the cluster as generic resources.

```APIDOC
## load_in_cluster_generic_resources

### Description
Loads all Custom Resource Definitions (CRDs) installed in the cluster as generic resources. This eliminates the need to explicitly define each resource type, which is particularly useful for scripting with unknown custom resources.

### Example
```python
from lightkube import Client
from lightkube.generic_resource import load_in_cluster_generic_resources

client = Client()
load_in_cluster_generic_resources(client)

# Now you can use load_all_yaml and interact with previously unknown resources.
```

### Note
This function is helpful when dealing with YAML files containing custom resources that are not pre-defined in lightkube.
```

--------------------------------

### Label Selector: Equal

Source: https://github.com/gtsystem/lightkube/blob/master/docs/selectors.md

Use this to match objects with a specific label key and value.

```python
labels={"env": "prod"}
```

--------------------------------

### Load Templated YAML Resources

Source: https://github.com/gtsystem/lightkube/blob/master/docs/codecs.md

Loads Kubernetes resource objects from a Jinja2 template file by providing a context dictionary for rendering. The rendered objects are returned as a list.

```python
with open('service.tmpl') as f:
    # render the template using `context` and return the corresponding resource objects.
    objs = codecs.load_all_yaml(f, context={'env': 'prd'})
    print(objs[0].metadata.labels['env'])  # prints `prd`
```

--------------------------------

### Sorting Objects for Delete

Source: https://github.com/gtsystem/lightkube/blob/master/docs/codecs.md

Illustrates how to use `sort_objects` with `reverse=True` to order Kubernetes resources for batch delete operations, preventing issues with dependent resources.

```APIDOC
Similarly, problems can arise when deleting a batch of objects. For example, consider the manifest `crs_and_crds.yaml`:

```yaml
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
...
spec:
  names:
    kind: SomeNewCr
    ...
---
kind: SomeNewCr
metadata:
  name: instance-of-new-cr
```

Deleting this in a loop like above would first delete the `CustomResourceDefinition`, resulting in all instances of `SomeNewCr` to be deleted implicitly. When we then attempted to delete `instance-of-new-cr`, we would encounter an API error. Use `codecs.sort_objects(..., reverse=True)` to avoid this issue:

```python
from lightkube import Client, codecs, sort_objects

client = Client()
with open('crs_amd_crds.yaml') as f:
    objects = codecs.load_all_yaml(f)
    for obj in sort_objects(objects, reverse=True):
        client.create(obj)
```

This orders the objects in a way that is friendly for deleting them as a batch.
```

--------------------------------

### Label Selector: Multiple Key-Value Pairs

Source: https://github.com/gtsystem/lightkube/blob/master/docs/selectors.md

Match objects that have all specified label key-value pairs.

```python
labels={"env": "prod", "app": "myapp"}
```

--------------------------------

### Apply Resources in Order with sort_objects

Source: https://github.com/gtsystem/lightkube/blob/master/docs/codecs.md

Use `sort_objects` to order resources for applying them sequentially. This prevents API errors caused by dependencies. Requires importing `Client`, `codecs`, and `sort_objects`.

```python
from lightkube import Client, codecs, sort_objects

client = Client()
with open('yaml_with_dependencies.yaml') as f:
    objects = codecs.load_all_yaml(f)
    for obj in sort_objects(objects):
        client.create(obj)
```

--------------------------------

### Async Loading In-Cluster Generic Resources

Source: https://github.com/gtsystem/lightkube/blob/master/docs/generic-resources.md

Introduces the asynchronous version of `load_in_cluster_generic_resources` for non-blocking operations.

```APIDOC
## async_load_in_cluster_generic_resources

### Description
Asynchronously loads all Custom Resource Definitions (CRDs) installed in the cluster as generic resources. This is the non-blocking counterpart to `load_in_cluster_generic_resources`.

### Method
`async_load_in_cluster_generic_resources(client: Client)`

### Note
Useful in asynchronous applications where blocking operations are undesirable.
```

--------------------------------

### Creating Generic Resources from CRD

Source: https://github.com/gtsystem/lightkube/blob/master/docs/generic-resources.md

Explains how to use `create_resources_from_crd` to generate generic resource definitions for each version of a `CustomResourceDefinition` object.

```APIDOC
## create_resources_from_crd

### Description
Creates generic resource definitions for each version specified in a `CustomResourceDefinition` object. This function is useful for programmatically generating resource types based on CRD definitions.

### Example
```python
from lightkube.generic_resource import create_resources_from_crd
from lightkube.resources.apiextensions_v1 import CustomResourceDefinition
from lightkube.models.apiextensions_v1 import (
    CustomResourceDefinitionNames,
    CustomResourceDefinitionSpec,
    CustomResourceDefinitionVersion,
)

versions = ['v1alpha1', 'v1']

crd = CustomResourceDefinition(
    spec=CustomResourceDefinitionSpec(
        group='some.group',
        names=CustomResourceDefinitionNames(
            kind='somekind',
            plural='somekinds',
        ),
        scope='Namespaced',
        versions=[CustomResourceDefinitionVersion(name=version, served=True, storage=True) for version in versions]
    )
)

create_resources_from_crd(crd)  # Creates generic resources for 'v1alpha1' and 'v1'

# To verify, you can inspect the internal registry:
from lightkube.generic_resource import _created_resources
print(_created_resources)
```
```

--------------------------------

### Add Label to ConfigMap

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Adds or updates a label on a ConfigMap resource. Requires 'lightkube'.

```python
client.set(ConfigMap, name="my-config", labels={'env': 'prod'})
```

--------------------------------

### Define Custom Resource Models

Source: https://github.com/gtsystem/lightkube/blob/master/docs/custom-resources.md

Define Python dataclasses that represent the structure of your custom resource, including its spec and status. These models should inherit from DictMixin.

```python
from typing import Optional

from lightkube.core.schema import DictMixin, dataclass
from lightkube.models import meta_v1


@dataclass
class Owner(DictMixin):
    name: str


@dataclass
class DogSpec(DictMixin):
    breed: str
    owner: Owner


@dataclass
class DogStatus(DictMixin):
    conditions: Optional[list[meta_v1.Condition]] = None
    observedGeneration: Optional[int] = None


@dataclass
class Dog(DictMixin):
    apiVersion: Optional[str] = None
    kind: Optional[str] = None
    metadata: Optional[meta_v1.ObjectMeta] = None
    spec: Optional[DogSpec] = None
    status: Optional[DogStatus] = None
```

--------------------------------

### Compare container request with limit

Source: https://github.com/gtsystem/lightkube/blob/master/docs/utils.md

Use `equals_canonically` to check if the requests and limits for a container are identical after being converted to their canonical forms. This ensures that minor formatting differences do not prevent an accurate equality check.

```python
from lightkube.utils.quantity import equals_canonically

pod = client.get(Pod, name='my-pod')
container_res = pod.spec.containers[0].resources
if equals_canonically(container_res.requests, container_res.limits):
    ... # requests and limits are the same ...
```

--------------------------------

### Set ConfigMap Label (Async)

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Asynchronously sets a label on a ConfigMap. Use when performing non-blocking operations.

```python
await client.set(ConfigMap, name="my-config", labels={'env': 'prod'})
```

--------------------------------

### Create Generic Resources from CRD Object

Source: https://github.com/gtsystem/lightkube/blob/master/docs/generic-resources.md

Define a CustomResourceDefinition object and use a helper function to create generic resource types for each of its versions. This makes these custom resources available for use with lightkube.

```python
from lightkube.generic_resource import create_resources_from_crd
from lightkube.resources.apiextensions_v1 import CustomResourceDefinition
from lightkube.models.apiextensions_v1 import (
    CustomResourceDefinitionNames,
    CustomResourceDefinitionSpec,
    CustomResourceDefinitionVersion,
)
versions = ['v1alpha1', 'v1']

crd = CustomResourceDefinition(
    
    spec=CustomResourceDefinitionSpec(
        group='some.group',
        names=CustomResourceDefinitionNames(
            kind='somekind',
            plural='somekinds',
        ),
        scope='Namespaced',
        versions=[
            CustomResourceDefinitionVersion(
                name=version,
                served=True,
                storage=True,
            ) for version in versions
        ],
    )
)

create_resources_from_crd(crd)  # Creates two generic resources, one for each above version

# To demonstrate this worked:
from lightkube.generic_resource import _created_resources
print("Dict of custom resources that have been defined in Lightkube:")
print(_created_resources)
```

--------------------------------

### Delete Namespaced Resource (Async)

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Asynchronously deletes a namespaced Kubernetes resource.

```python
await client.delete(ConfigMap, name='my-config', namespace='default')
```

--------------------------------

### Patch ConfigMap with New Data

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Adds new data to an existing ConfigMap using a patch operation. Requires 'lightkube'.

```python
patch = {"data": {"key3": "value3"}}
client.patch(ConfigMap, name="my-config", obj=patch)
```

--------------------------------

### Delete Resources in Reverse Order with sort_objects

Source: https://github.com/gtsystem/lightkube/blob/master/docs/codecs.md

Use `sort_objects(..., reverse=True)` to order resources for deleting them sequentially. This is useful for CRDs and their instances to avoid errors. Requires importing `Client`, `codecs`, and `sort_objects`.

```python
from lightkube import Client, codecs, sort_objects

client = Client()
with open('crs_amd_crds.yaml') as f:
    objects = codecs.load_all_yaml(f)
    for obj in sort_objects(objects, reverse=True):
        client.create(obj)
```

--------------------------------

### Label Selector: Key Exists

Source: https://github.com/gtsystem/lightkube/blob/master/docs/selectors.md

Match objects that have a specific label key, regardless of its value.

```python
labels={"env": "prod", "app": None}
```

--------------------------------

### Access Generic Resource Content

Source: https://github.com/gtsystem/lightkube/blob/master/docs/generic-resources.md

Access the content of a generic resource using dictionary-like or attribute notation. Metadata is decoded into a specific model.

```python
print(job["path"]["to"]["something"])
```

```python
print(job.kind)
print(job.metadata)
```

```python
print(job.metadata.name)
```

--------------------------------

### Label Selector: Not Exists

Source: https://github.com/gtsystem/lightkube/blob/master/docs/selectors.md

Match objects that do not have a specific label key.

```python
labels={"app": op.not_exists()}
```

--------------------------------

### Delete a Namespaced Resource

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Deletes a specific ConfigMap from the 'default' namespace. Assumes 'client' is defined.

```python
client.delete(ConfigMap, name='my-config', namespace='default')
```

--------------------------------

### Compare container memory request with limit

Source: https://github.com/gtsystem/lightkube/blob/master/docs/utils.md

Use `parse_quantity` to convert memory requests and limits to a comparable decimal format. This is useful for validating that a container's request is less than its limit.

```python
from lightkube.utils.quantity import parse_quantity

pod = client.get(Pod, name='my-pod')
container_res = pod.spec.containers[0].resources
if parse_quantity(container_res.requests['memory']) < parse_quantity(container_res.limits['memory']):
    ... # request is less than limit, do something ...
```

--------------------------------

### Replace a ConfigMap asynchronously

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Replace an existing ConfigMap with modified data using the asynchronous client. Assumes 'config' object is already defined.

```python
config.data['key1'] = 'new value'
await client.replace(config)
```

--------------------------------

### Label Selector: In (OR condition)

Source: https://github.com/gtsystem/lightkube/blob/master/docs/selectors.md

Match objects where a label's value is one of the specified values.

```python
labels={"env": ("prod", "dev")}
```

--------------------------------

### Patch a ConfigMap asynchronously

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Patch an existing ConfigMap to add new data using the asynchronous client. Requires the ConfigMap name and namespace.

```python
patch = {"data": {"key3": "value3"}}
await client.patch(ConfigMap, name='my-config', obj=patch)
```

--------------------------------

### Handle Lightkube API Errors

Source: https://github.com/gtsystem/lightkube/blob/master/docs/reference/exceptions.md

Catch ApiError for HTTP errors returned from the Kubernetes API. The status attribute provides details about the failure.

```python
from lightkube import Client, ApiError

client = Client()
try:
    pod = client.get(Pod, name="not-existing-pod")
except ApiError as e:
    print(e.status)
```

--------------------------------

### Label Selector: Not Equal

Source: https://github.com/gtsystem/lightkube/blob/master/docs/selectors.md

Match objects where a label's value is not equal to the specified value.

```python
labels={"env": op.not_equal("prod")}
```

--------------------------------

### Remove Label from ConfigMap

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Removes a label from a ConfigMap resource by setting its value to None. Requires 'lightkube'.

```python
client.set(ConfigMap, name="my-config", labels={'env': None})
```

--------------------------------

### Replace ConfigMap Data

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Modifies an existing ConfigMap by changing its data content. Assumes 'config' object and 'client' are already defined.

```python
config.data['key1'] = 'new value'
client.replace(config)
```

--------------------------------

### Remove Data Key from ConfigMap using Patch

Source: https://github.com/gtsystem/lightkube/blob/master/README.md

Removes a specific data key from a ConfigMap using a merge patch. Setting the value to None effectively removes the key. Requires 'lightkube' and 'PatchType.MERGE'.

```python
# When using PatchType.MERGE, setting a value of a key/value to None, will remove the current item 
patch = {'metadata': {"key3": None}}
client.patch(ConfigMap, name='my-config', namespace='default', obj=patch, patch_type=PatchType.MERGE)
```

--------------------------------

### Field Selector: Not in Namespace

Source: https://github.com/gtsystem/lightkube/blob/master/docs/selectors.md

Select resources that are not in a specific namespace. Note that only 'equal', 'not equal', and 'not in' operations are supported for field selectors.

```python
fields={"metadata.namespace": op.not_equal("default")}
```

--------------------------------

### Label Selector: Not In

Source: https://github.com/gtsystem/lightkube/blob/master/docs/selectors.md

Match objects where a label's value is not among the specified values.

```python
labels={"env": op.not_in(["prod", "dev"])}
```

--------------------------------

### Remove data from ConfigMap asynchronously

Source: https://github.com/gtsystem/lightkube/blob/master/docs/index.md

Remove a specific data key from a ConfigMap using a merge patch where the value is set to None. Requires ConfigMap name, namespace, and patch type.

```python
# When using PatchType.MERGE, setting a value of a key/value to None, will remove the current item 
patch = {'metadata': {"key3": None}}
await client.patch(ConfigMap, name='my-config', namespace='default', obj=patch, patch_type=PatchType.MERGE)
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.