custom python kubernetes client

k8s_client.py

from collections.abc import AsyncIterator
from dataclasses import dataclass, field

from httpx_util import BearerAuth, async_httpx

# https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md
# https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/BatchV1Api.md


@dataclass(frozen=True)
class KubernetesClient:
    host: str
    namespace: str
    api_token: str = field(repr=False)

    async def list_pods(self) -> AsyncIterator[str]:
        url = f'https://{self.host}/api/v1/namespaces/{self.namespace}/pods'
        response = await async_httpx(method='get', url=url, auth=BearerAuth(self.api_token), verify=False)
        response.raise_for_status()
        response_data = response.json()
        for item in response_data['items']:
            yield item['metadata']['name']


if __name__ == '__main__':
    import asyncio
    import os

    kubernetes_host: str = os.environ['KUBERNETES_HOST']
    kubernetes_namespace: str = os.environ['KUBERNETES_NAMESPACE']
    kubernetes_token: str = os.environ['KUBERNETES_TOKEN']

    k8s_client = KubernetesClient(host=kubernetes_host, namespace=kubernetes_namespace, api_token=kubernetes_token)

    async def test_list_pods():
        async for name in k8s_client.list_pods():
            print(name)

    asyncio.run(test_list_pods())