0
# Kubernetes Asyncio
1

2
Asynchronous Python client library for the Kubernetes API providing async/await support for all Kubernetes operations. This package enables non-blocking interactions with Kubernetes clusters from Python applications, built using the same OpenAPI generator approach as the official Kubernetes Python client but with full asynchronous support.
3

4
## Package Information
5

6
- **Package Name**: kubernetes_asyncio
7
- **Language**: Python
8
- **Installation**: `pip install kubernetes_asyncio`
9
- **OpenAPI Version**: v1.33.3
10
- **Package Version**: 33.3.0
11

12
## Core Imports
13

14
```python
15
import kubernetes_asyncio
16
```
17

18
Common imports for basic usage:
19

20
```python
21
from kubernetes_asyncio import client, config
22
```
23

24
Dynamic client import:
25

26
```python
27
from kubernetes_asyncio import dynamic
28
```
29

30
Streaming and watch functionality:
31

32
```python
33
from kubernetes_asyncio import stream, watch
34
```
35

36
Utility functions:
37

38
```python
39
from kubernetes_asyncio import utils
40
```
41

42
Leader election:
43

44
```python
45
from kubernetes_asyncio.leaderelection import leaderelection, electionconfig
46
from kubernetes_asyncio.leaderelection.resourcelock import leaselock, configmaplock
47
```
48

49
## Basic Usage
50

51
```python
52
import asyncio
53
from kubernetes_asyncio import client, config
54

55
async def main():
56
    # Load configuration (tries kubeconfig, then in-cluster)
57
    await config.load_config()
58
    
59
    # Use context manager to ensure proper cleanup
60
    async with client.ApiClient() as api:
61
        v1 = client.CoreV1Api(api)
62
        
63
        # List all pods in default namespace
64
        pod_list = await v1.list_namespaced_pod(namespace="default")
65
        for pod in pod_list.items:
66
            print(f"Pod: {pod.metadata.name}")
67
        
68
        # Create a simple pod
69
        pod_manifest = {
70
            "apiVersion": "v1",
71
            "kind": "Pod",
72
            "metadata": {"name": "test-pod"},
73
            "spec": {
74
                "containers": [{
75
                    "name": "test-container",
76
                    "image": "nginx:latest",
77
                    "ports": [{"containerPort": 80}]
78
                }]
79
            }
80
        }
81
        
82
        pod = client.V1Pod(**pod_manifest)
83
        await v1.create_namespaced_pod(namespace="default", body=pod)
84
        print("Pod created successfully")
85

86
# Run the async function
87
asyncio.run(main())
88
```
89

90
## Architecture
91

92
The kubernetes_asyncio library follows a modular architecture that mirrors the official Kubernetes Python client:
93

94
- **Client Module**: Auto-generated API classes (64 total) corresponding to all Kubernetes API groups and versions, plus 694 model classes representing all Kubernetes resource types
95
- **Configuration Module**: Multiple authentication methods including kubeconfig files, in-cluster configuration, exec providers, and OIDC
96
- **Dynamic Client**: Runtime API discovery allowing interaction with any Kubernetes resource, including custom resources
97
- **Streaming**: WebSocket support for exec sessions and port forwarding operations  
98
- **Watch**: Event streaming for monitoring resource changes in real-time
99
- **Utils**: YAML processing utilities for applying manifests from files or dictionaries
100
- **Leader Election**: Distributed coordination using ConfigMaps or Lease resources
101

102
All operations use Python's async/await pattern with proper resource cleanup through context managers.
103

104
## Capabilities
105

106
### Client APIs
107

108
Core Kubernetes API access through 64 auto-generated API classes covering all Kubernetes API groups and versions. Includes comprehensive model classes for all resource types with full async/await support.
109

110
```python { .api }
111
class CoreV1Api:
112
    async def list_pod_for_all_namespaces(**kwargs): ...
113
    async def list_namespaced_pod(namespace: str, **kwargs): ...
114
    async def create_namespaced_pod(namespace: str, body: V1Pod, **kwargs): ...
115
    async def read_namespaced_pod(name: str, namespace: str, **kwargs): ...
116
    async def patch_namespaced_pod(name: str, namespace: str, body: object, **kwargs): ...
117
    async def delete_namespaced_pod(name: str, namespace: str, **kwargs): ...
118

119
class AppsV1Api:
120
    async def list_namespaced_deployment(namespace: str, **kwargs): ...
121
    async def create_namespaced_deployment(namespace: str, body: V1Deployment, **kwargs): ...
122
    async def read_namespaced_deployment(name: str, namespace: str, **kwargs): ...
123
    async def patch_namespaced_deployment(name: str, namespace: str, body: object, **kwargs): ...
124
    async def delete_namespaced_deployment(name: str, namespace: str, **kwargs): ...
125

126
class ApiClient:
127
    async def __aenter__(self): ...
128
    async def __aexit__(self, exc_type, exc_val, exc_tb): ...
129
    async def close(self): ...
130
```
131

132
[Client APIs](./client-apis.md)
133

134
### Configuration
135

136
Multiple methods for loading Kubernetes cluster credentials and configuration, supporting kubeconfig files, in-cluster authentication, and various credential providers.
137

138
```python { .api }
139
async def load_config(**kwargs):
140
    """
141
    Load Kubernetes configuration from kubeconfig file or in-cluster.
142
    
143
    Parameters:
144
    - config_file: str, path to kubeconfig file
145
    - context: str, kubeconfig context to use
146
    - client_configuration: Configuration object to populate
147
    - persist_config: bool, whether to persist config globally
148
    """
149

150
def load_incluster_config(**kwargs):
151
    """Load in-cluster configuration from service account."""
152

153
async def load_kube_config(config_file=None, context=None, **kwargs):
154
    """Load configuration from kubeconfig file."""
155

156
def new_client_from_config(**kwargs) -> ApiClient:
157
    """Create new ApiClient from configuration."""
158
```
159

160
[Configuration](./configuration.md)
161

162
### Dynamic Client
163

164
Runtime API discovery and interaction with any Kubernetes resource, including custom resources. Enables working with resources not known at compile time.
165

166
```python { .api }
167
class DynamicClient:
168
    def __init__(self, client: ApiClient, cache_file=None, discoverer=None): ...
169
    async def __aenter__(self): ...
170
    async def __aexit__(self, exc_type, exc_val, exc_tb): ...
171
    @property
172
    def resources(self): ...
173
    
174
class Resource:
175
    def get(self, name=None, namespace=None, **kwargs): ...
176
    def create(self, body=None, namespace=None, **kwargs): ...
177
    def patch(self, body=None, name=None, namespace=None, **kwargs): ...
178
    def delete(self, name=None, namespace=None, **kwargs): ...
179
```
180

181
[Dynamic Client](./dynamic-client.md)
182

183
### Streaming
184

185
WebSocket-based streaming for exec sessions, port forwarding, and attach operations. Provides real-time bidirectional communication with containers.
186

187
```python { .api }
188
class WsApiClient(ApiClient):
189
    def __init__(self, configuration=None, header_name=None, header_value=None, 
190
                 cookie=None, pool_threads=1, heartbeat=None): ...
191
    async def request(self, method, url, **kwargs): ...
192

193
# Channel constants
194
STDIN_CHANNEL: int = 0
195
STDOUT_CHANNEL: int = 1  
196
STDERR_CHANNEL: int = 2
197
ERROR_CHANNEL: int = 3
198
RESIZE_CHANNEL: int = 4
199
```
200

201
[Streaming](./streaming.md)
202

203
### Watch
204

205
Event streaming for monitoring resource changes in real-time. Efficiently tracks additions, modifications, deletions, and errors for any Kubernetes resource.
206

207
```python { .api }
208
class Watch:
209
    def __init__(self, return_type=None): ...
210
    def stop(self): ...
211
    async def stream(self, func, *args, **kwargs):
212
        """
213
        Stream events from a Kubernetes API list operation.
214
        
215
        Parameters:
216
        - func: API function that supports watch parameter
217
        - *args, **kwargs: Arguments passed to the API function
218
        
219
        Yields:
220
        - dict: Event with 'type' ('ADDED', 'MODIFIED', 'DELETED', 'ERROR') and 'object'
221
        """
222
```
223

224
[Watch](./watch.md)
225

226
### Utils
227

228
YAML processing utilities for creating and applying Kubernetes resources from YAML files or dictionaries with full async support.
229

230
```python { .api }
231
async def create_from_yaml(k8s_client: ApiClient, yaml_file: str, 
232
                          verbose: bool = False, namespace: str = "default", **kwargs):
233
    """
234
    Create Kubernetes resources from YAML file.
235
    
236
    Parameters:
237
    - k8s_client: Kubernetes API client
238
    - yaml_file: Path to YAML file containing resource definitions
239
    - verbose: Enable verbose output
240
    - namespace: Default namespace for resources
241
    - **kwargs: Additional arguments passed to create operations
242
    """
243

244
async def create_from_dict(k8s_client: ApiClient, data: dict,
245
                          verbose: bool = False, namespace: str = "default", **kwargs):
246
    """Create Kubernetes resources from dictionary."""
247

248
class FailToCreateError(Exception):
249
    """Exception raised when resource creation fails."""
250
```
251

252
[Utils](./utils.md)
253

254
### Leader Election
255

256
Distributed leader election for high availability applications using Kubernetes native resources (ConfigMaps or Leases) for coordination.
257

258
```python { .api }
259
class LeaderElection:
260
    def __init__(self, election_config: ElectionConfig): ...
261
    async def run(self): ...
262
    async def acquire(self): ...
263
    async def renew_loop(self): ...
264

265
class ElectionConfig:
266
    def __init__(self, lock, lease_duration, renew_deadline, retry_period, 
267
                 onstarted_leading=None, onstopped_leading=None): ...
268

269
class LeaseLock:
270
    def __init__(self, name: str, namespace: str, identity: str): ...
271

272
class ConfigMapLock:
273
    def __init__(self, name: str, namespace: str, identity: str): ...
274
```
275

276
[Leader Election](./leader-election.md)
277

278
## Error Handling
279

280
The library provides comprehensive exception handling with specific exception types for different error conditions:
281

282
```python { .api }
283
class ApiException(Exception):
284
    """General API errors with HTTP status, reason, and response body."""
285
    def __init__(self, status=None, reason=None, http_resp=None): ...
286

287
class ApiTypeError(TypeError):
288
    """Type validation errors."""
289

290
class ApiValueError(ValueError):
291
    """Value validation errors."""
292

293
class ConfigException(Exception):
294
    """Configuration-related errors."""
295

296
class FailToCreateError(Exception):
297
    """Resource creation failures."""
298
```
299

300
Common error handling pattern:
301

302
```python
303
try:
304
    result = await v1.create_namespaced_pod(namespace="default", body=pod)
305
except client.ApiException as e:
306
    if e.status == 409:
307
        print("Pod already exists")
308
    elif e.status == 403:
309
        print("Access denied")
310
    else:
311
        print(f"API error: {e.status} {e.reason}")
312
except Exception as e:
313
    print(f"Unexpected error: {e}")
314
```