Kubernetes Asynchronous Python Client
kubernetes-asyncio is an asynchronous (AsyncIO) client library for interacting with the Kubernetes API. It provides non-blocking access to Kubernetes clusters from Python applications, built using the same OpenAPI generator approach as the official Kubernetes Python client, but with full async/await support. The library is actively maintained, with releases frequently aligning with the Kubernetes project's own approximately three-times-a-year release cadence.
Warnings
- breaking The versioning scheme changed significantly with `v18.20.0`. The first two numbers of the library version now indicate the target Kubernetes API version (e.g., `18.20.x` for Kubernetes `v1.18.x`). This is a departure from standard semantic versioning for the library itself and can cause confusion regarding compatibility.
- gotcha On Microsoft Windows, the default `asyncio.SelectorEventLoop` might not support pipes and subprocesses, causing `exec_provider.py::ExecProvider` (used for client-go credentials plugins) to fail. This is a common footgun for Windows users.
- gotcha It is crucial to use `ApiClient` within an `async with` statement (i.e., as a context manager) to ensure that underlying HTTP sessions are properly closed and resources are released. Failing to do so can lead to resource leaks or unexpected behavior in long-running applications.
Install
-
pip install kubernetes_asyncio
Imports
- client, config
from kubernetes_asyncio import client, config
- ApiClient
from kubernetes_asyncio.client.api_client import ApiClient
Quickstart
import asyncio
from kubernetes_asyncio import client, config
from kubernetes_asyncio.client.api_client import ApiClient
async def main():
# Configs can be loaded from default locations (e.g., ~/.kube/config)
# or explicitly from a file, or in-cluster configuration.
# No argument provided means it will try default locations.
await config.load_kube_config()
# Use the context manager to ensure http sessions are closed automatically
async with ApiClient() as api:
v1 = client.CoreV1Api(api)
print("Listing pods with their IPs:")
ret = await v1.list_pod_for_all_namespaces()
for i in ret.items:
print(f"{i.status.pod_ip}\t{i.metadata.namespace}\t{i.metadata.name}")
if __name__ == '__main__':
asyncio.run(main())