Skip to content

prefect_kubernetes.pods

Module for interacting with Kubernetes pods from Prefect flows.

create_namespaced_pod(kubernetes_credentials, new_pod, namespace='default', **kube_kwargs) async

Create a Kubernetes pod in a given namespace.

Parameters:

Name Type Description Default
kubernetes_credentials KubernetesCredentials

KubernetesCredentials block for creating authenticated Kubernetes API clients.

 required
new_pod V1Pod

A Kubernetes V1Pod specification.

 required
namespace Optional[str]

The Kubernetes namespace to create this pod in.

 'default'
**kube_kwargs Dict[str, Any]

Optional extra keyword arguments to pass to the Kubernetes API.

 {}

Returns:

Type Description
V1Pod

A Kubernetes V1Pod object.
Example

Create a pod in the default namespace:

from prefect import flow
from prefect_kubernetes.credentials import KubernetesCredentials
from prefect_kubernetes.pods import create_namespaced_pod
from kubernetes_asyncio.client.models import V1Pod

@flow
def kubernetes_orchestrator():
    v1_pod_metadata = create_namespaced_pod(
        kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
        new_pod=V1Pod(metadata={"name": "test-pod"}),
    )
Source code in prefect_kubernetes/pods.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@task
async def create_namespaced_pod(
    kubernetes_credentials: KubernetesCredentials,
    new_pod: V1Pod,
    namespace: Optional[str] = "default",
    **kube_kwargs: Dict[str, Any],
) -> V1Pod:
    """Create a Kubernetes pod in a given namespace.

    Args:
        kubernetes_credentials: `KubernetesCredentials` block for creating
            authenticated Kubernetes API clients.
        new_pod: A Kubernetes `V1Pod` specification.
        namespace: The Kubernetes namespace to create this pod in.
        **kube_kwargs: Optional extra keyword arguments to pass to the Kubernetes API.

    Returns:
        A Kubernetes `V1Pod` object.

    Example:
        Create a pod in the default namespace:
        ```python
        from prefect import flow
        from prefect_kubernetes.credentials import KubernetesCredentials
        from prefect_kubernetes.pods import create_namespaced_pod
        from kubernetes_asyncio.client.models import V1Pod

        @flow
        def kubernetes_orchestrator():
            v1_pod_metadata = create_namespaced_pod(
                kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
                new_pod=V1Pod(metadata={"name": "test-pod"}),
            )
        ```
    """
    async with kubernetes_credentials.get_client("core") as core_v1_client:
        return await core_v1_client.create_namespaced_pod(
            namespace=namespace,
            body=new_pod,
            **kube_kwargs,
        )

delete_namespaced_pod(kubernetes_credentials, pod_name, delete_options=None, namespace='default', **kube_kwargs) async

Delete a Kubernetes pod in a given namespace.

Parameters:

Name Type Description Default
kubernetes_credentials KubernetesCredentials

KubernetesCredentials block for creating authenticated Kubernetes API clients.

 required
pod_name str

The name of the pod to delete.

 required
delete_options Optional[V1DeleteOptions]

A Kubernetes V1DeleteOptions object.

 None
namespace Optional[str]

The Kubernetes namespace to delete this pod from.

 'default'
**kube_kwargs Dict[str, Any]

Optional extra keyword arguments to pass to the Kubernetes API.

 {}

Returns:

Type Description
V1Pod

A Kubernetes V1Pod object.
Example

Delete a pod in the default namespace:

from prefect import flow
from prefect_kubernetes.credentials import KubernetesCredentials
from prefect_kubernetes.pods import delete_namespaced_pod
from kubernetes_asyncio.client.models import V1DeleteOptions

@flow
def kubernetes_orchestrator():
    v1_pod_metadata = delete_namespaced_pod(
        kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
        pod_name="test-pod",
        delete_options=V1DeleteOptions(grace_period_seconds=0),
    )
Source code in prefect_kubernetes/pods.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@task
async def delete_namespaced_pod(
    kubernetes_credentials: KubernetesCredentials,
    pod_name: str,
    delete_options: Optional[V1DeleteOptions] = None,
    namespace: Optional[str] = "default",
    **kube_kwargs: Dict[str, Any],
) -> V1Pod:
    """Delete a Kubernetes pod in a given namespace.

    Args:
        kubernetes_credentials: `KubernetesCredentials` block for creating
            authenticated Kubernetes API clients.
        pod_name: The name of the pod to delete.
        delete_options: A Kubernetes `V1DeleteOptions` object.
        namespace: The Kubernetes namespace to delete this pod from.
        **kube_kwargs: Optional extra keyword arguments to pass to the Kubernetes API.

    Returns:
        A Kubernetes `V1Pod` object.

    Example:
        Delete a pod in the default namespace:
        ```python
        from prefect import flow
        from prefect_kubernetes.credentials import KubernetesCredentials
        from prefect_kubernetes.pods import delete_namespaced_pod
        from kubernetes_asyncio.client.models import V1DeleteOptions

        @flow
        def kubernetes_orchestrator():
            v1_pod_metadata = delete_namespaced_pod(
                kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
                pod_name="test-pod",
                delete_options=V1DeleteOptions(grace_period_seconds=0),
            )
        ```
    """
    async with kubernetes_credentials.get_client("core") as core_v1_client:
        return await core_v1_client.delete_namespaced_pod(
            pod_name,
            body=delete_options,
            namespace=namespace,
            **kube_kwargs,
        )

list_namespaced_pod(kubernetes_credentials, namespace='default', **kube_kwargs) async

List all pods in a given namespace.

Parameters:

Name Type Description Default
kubernetes_credentials KubernetesCredentials

KubernetesCredentials block for creating authenticated Kubernetes API clients.

 required
namespace Optional[str]

The Kubernetes namespace to list pods from.

 'default'
**kube_kwargs Dict[str, Any]

Optional extra keyword arguments to pass to the Kubernetes API.

 {}

Returns:

Type Description
V1PodList

A Kubernetes V1PodList object.
Example

List all pods in the default namespace:

from prefect import flow
from prefect_kubernetes.credentials import KubernetesCredentials
from prefect_kubernetes.pods import list_namespaced_pod

@flow
def kubernetes_orchestrator():
    v1_pod_list = list_namespaced_pod(
        kubernetes_credentials=KubernetesCredentials.load("k8s-creds")
    )
Source code in prefect_kubernetes/pods.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
@task
async def list_namespaced_pod(
    kubernetes_credentials: KubernetesCredentials,
    namespace: Optional[str] = "default",
    **kube_kwargs: Dict[str, Any],
) -> V1PodList:
    """List all pods in a given namespace.

    Args:
        kubernetes_credentials: `KubernetesCredentials` block for creating
            authenticated Kubernetes API clients.
        namespace: The Kubernetes namespace to list pods from.
        **kube_kwargs: Optional extra keyword arguments to pass to the Kubernetes API.

    Returns:
        A Kubernetes `V1PodList` object.

    Example:
        List all pods in the default namespace:
        ```python
        from prefect import flow
        from prefect_kubernetes.credentials import KubernetesCredentials
        from prefect_kubernetes.pods import list_namespaced_pod

        @flow
        def kubernetes_orchestrator():
            v1_pod_list = list_namespaced_pod(
                kubernetes_credentials=KubernetesCredentials.load("k8s-creds")
            )
        ```
    """
    async with kubernetes_credentials.get_client("core") as core_v1_client:
        return await core_v1_client.list_namespaced_pod(
            namespace=namespace, **kube_kwargs
        )

patch_namespaced_pod(kubernetes_credentials, pod_name, pod_updates, namespace='default', **kube_kwargs) async

Patch a Kubernetes pod in a given namespace.

Parameters:

Name Type Description Default
kubernetes_credentials KubernetesCredentials

KubernetesCredentials block for creating authenticated Kubernetes API clients.

 required
pod_name str

The name of the pod to patch.

 required
pod_updates V1Pod

A Kubernetes V1Pod object.

 required
namespace Optional[str]

The Kubernetes namespace to patch this pod in.

 'default'
**kube_kwargs Dict[str, Any]

Optional extra keyword arguments to pass to the Kubernetes API.

 {}

Returns:

Type Description
V1Pod

A Kubernetes V1Pod object.
Example

Patch a pod in the default namespace:

from prefect import flow
from prefect_kubernetes.credentials import KubernetesCredentials
from prefect_kubernetes.pods import patch_namespaced_pod
from kubernetes_asyncio.client.models import V1Pod

@flow
def kubernetes_orchestrator():
    v1_pod_metadata = patch_namespaced_pod(
        kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
        pod_name="test-pod",
        pod_updates=V1Pod(metadata={"labels": {"foo": "bar"}}),
    )
Source code in prefect_kubernetes/pods.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
@task
async def patch_namespaced_pod(
    kubernetes_credentials: KubernetesCredentials,
    pod_name: str,
    pod_updates: V1Pod,
    namespace: Optional[str] = "default",
    **kube_kwargs: Dict[str, Any],
) -> V1Pod:
    """Patch a Kubernetes pod in a given namespace.

    Args:
        kubernetes_credentials: `KubernetesCredentials` block for creating
            authenticated Kubernetes API clients.
        pod_name: The name of the pod to patch.
        pod_updates: A Kubernetes `V1Pod` object.
        namespace: The Kubernetes namespace to patch this pod in.
        **kube_kwargs: Optional extra keyword arguments to pass to the Kubernetes API.

    Returns:
        A Kubernetes `V1Pod` object.

    Example:
        Patch a pod in the default namespace:
        ```python
        from prefect import flow
        from prefect_kubernetes.credentials import KubernetesCredentials
        from prefect_kubernetes.pods import patch_namespaced_pod
        from kubernetes_asyncio.client.models import V1Pod

        @flow
        def kubernetes_orchestrator():
            v1_pod_metadata = patch_namespaced_pod(
                kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
                pod_name="test-pod",
                pod_updates=V1Pod(metadata={"labels": {"foo": "bar"}}),
            )
        ```
    """
    async with kubernetes_credentials.get_client("core") as core_v1_client:
        return await core_v1_client.patch_namespaced_pod(
            name=pod_name,
            namespace=namespace,
            body=pod_updates,
            **kube_kwargs,
        )

read_namespaced_pod(kubernetes_credentials, pod_name, namespace='default', **kube_kwargs) async

Read information on a Kubernetes pod in a given namespace.

Parameters:

Name Type Description Default
kubernetes_credentials KubernetesCredentials

KubernetesCredentials block for creating authenticated Kubernetes API clients.

 required
pod_name str

The name of the pod to read.

 required
namespace Optional[str]

The Kubernetes namespace to read this pod from.

 'default'
**kube_kwargs Dict[str, Any]

Optional extra keyword arguments to pass to the Kubernetes API.

 {}

Returns:

Type Description
V1Pod

A Kubernetes V1Pod object.
Example

Read a pod in the default namespace:

from prefect import flow
from prefect_kubernetes.credentials import KubernetesCredentials

@flow
def kubernetes_orchestrator():
    v1_pod_metadata = read_namespaced_pod(
        kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
        pod_name="test-pod"
    )
Source code in prefect_kubernetes/pods.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
@task
async def read_namespaced_pod(
    kubernetes_credentials: KubernetesCredentials,
    pod_name: str,
    namespace: Optional[str] = "default",
    **kube_kwargs: Dict[str, Any],
) -> V1Pod:
    """Read information on a Kubernetes pod in a given namespace.

    Args:
        kubernetes_credentials: `KubernetesCredentials` block for creating
            authenticated Kubernetes API clients.
        pod_name: The name of the pod to read.
        namespace: The Kubernetes namespace to read this pod from.
        **kube_kwargs: Optional extra keyword arguments to pass to the Kubernetes API.

    Returns:
        A Kubernetes `V1Pod` object.

    Example:
        Read a pod in the default namespace:
        ```python
        from prefect import flow
        from prefect_kubernetes.credentials import KubernetesCredentials

        @flow
        def kubernetes_orchestrator():
            v1_pod_metadata = read_namespaced_pod(
                kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
                pod_name="test-pod"
            )
        ```
    """
    async with kubernetes_credentials.get_client("core") as core_v1_client:
        return await core_v1_client.read_namespaced_pod(
            name=pod_name,
            namespace=namespace,
            **kube_kwargs,
        )

read_namespaced_pod_log(kubernetes_credentials, pod_name, container, namespace='default', print_func=None, **kube_kwargs) async

Read logs from a Kubernetes pod in a given namespace.

If print_func is provided, the logs will be streamed using that function. If the pod is no longer running, logs generated up to that point will be returned.

Parameters:

Name Type Description Default
kubernetes_credentials KubernetesCredentials

KubernetesCredentials block for creating authenticated Kubernetes API clients.

 required
pod_name str

The name of the pod to read logs from.

 required
container str

The name of the container to read logs from.

 required
namespace Optional[str]

The Kubernetes namespace to read this pod from.

 'default'
print_func Optional[Callable]

If provided, it will stream the pod logs by calling print_func for every line and returning None. If not provided, the current pod logs will be returned immediately.

 None
**kube_kwargs Dict[str, Any]

Optional extra keyword arguments to pass to the Kubernetes API.

 {}

Returns:

Type Description
Union[str, None]

A string containing the logs from the pod's container.
Example

Read logs from a pod in the default namespace:

from prefect import flow
from prefect.logging import get_run_logger
from prefect_kubernetes.credentials import KubernetesCredentials
from prefect_kubernetes.pods import read_namespaced_pod_logs

@flow
def kubernetes_orchestrator():
    logger = get_run_logger()

    pod_logs = read_namespaced_pod_logs(
        kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
        pod_name="test-pod",
        container="test-container",
        print_func=logger.info
    )
Source code in prefect_kubernetes/pods.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
@task
async def read_namespaced_pod_log(
    kubernetes_credentials: KubernetesCredentials,
    pod_name: str,
    container: str,
    namespace: Optional[str] = "default",
    print_func: Optional[Callable] = None,
    **kube_kwargs: Dict[str, Any],
) -> Union[str, None]:
    """Read logs from a Kubernetes pod in a given namespace.

    If `print_func` is provided, the logs will be streamed using that function.
    If the pod is no longer running, logs generated up to that point will be returned.

    Args:
        kubernetes_credentials: `KubernetesCredentials` block for creating
            authenticated Kubernetes API clients.
        pod_name: The name of the pod to read logs from.
        container: The name of the container to read logs from.
        namespace: The Kubernetes namespace to read this pod from.
        print_func: If provided, it will stream the pod logs by calling `print_func`
            for every line and returning `None`. If not provided, the current pod
            logs will be returned immediately.
        **kube_kwargs: Optional extra keyword arguments to pass to the Kubernetes API.

    Returns:
        A string containing the logs from the pod's container.

    Example:
        Read logs from a pod in the default namespace:
        ```python
        from prefect import flow
        from prefect.logging import get_run_logger
        from prefect_kubernetes.credentials import KubernetesCredentials
        from prefect_kubernetes.pods import read_namespaced_pod_logs

        @flow
        def kubernetes_orchestrator():
            logger = get_run_logger()

            pod_logs = read_namespaced_pod_logs(
                kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
                pod_name="test-pod",
                container="test-container",
                print_func=logger.info
            )
        ```
    """
    async with kubernetes_credentials.get_client("core") as core_v1_client:
        if print_func is not None:
            # should no longer need to manually refresh on ApiException.status == 410
            # as of https://github.com/kubernetes-client/python-base/pull/133
            async with Watch() as watch:
                async for log_line in watch.stream(
                    core_v1_client.read_namespaced_pod_log,
                    name=pod_name,
                    namespace=namespace,
                    container=container,
                ):
                    print_func(log_line)

        return await core_v1_client.read_namespaced_pod_log(
            name=pod_name,
            namespace=namespace,
            container=container,
            **kube_kwargs,
        )

replace_namespaced_pod(kubernetes_credentials, pod_name, new_pod, namespace='default', **kube_kwargs) async

Replace a Kubernetes pod in a given namespace.

Parameters:

Name Type Description Default
kubernetes_credentials KubernetesCredentials

KubernetesCredentials block for creating authenticated Kubernetes API clients.

 required
pod_name str

The name of the pod to replace.

 required
new_pod V1Pod

A Kubernetes V1Pod object.

 required
namespace Optional[str]

The Kubernetes namespace to replace this pod in.

 'default'
**kube_kwargs Dict[str, Any]

Optional extra keyword arguments to pass to the Kubernetes API.

 {}

Returns:

Type Description
V1Pod

A Kubernetes V1Pod object.
Example

Replace a pod in the default namespace:

from prefect import flow
from prefect_kubernetes.credentials import KubernetesCredentials
from prefect_kubernetes.pods import replace_namespaced_pod
from kubernetes_asyncio.client.models import V1Pod

@flow
def kubernetes_orchestrator():
    v1_pod_metadata = replace_namespaced_pod(
        kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
        pod_name="test-pod",
        new_pod=V1Pod(metadata={"labels": {"foo": "bar"}})
    )
Source code in prefect_kubernetes/pods.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
@task
async def replace_namespaced_pod(
    kubernetes_credentials: KubernetesCredentials,
    pod_name: str,
    new_pod: V1Pod,
    namespace: Optional[str] = "default",
    **kube_kwargs: Dict[str, Any],
) -> V1Pod:
    """Replace a Kubernetes pod in a given namespace.

    Args:
        kubernetes_credentials: `KubernetesCredentials` block for creating
            authenticated Kubernetes API clients.
        pod_name: The name of the pod to replace.
        new_pod: A Kubernetes `V1Pod` object.
        namespace: The Kubernetes namespace to replace this pod in.
        **kube_kwargs: Optional extra keyword arguments to pass to the Kubernetes API.

    Returns:
        A Kubernetes `V1Pod` object.

    Example:
        Replace a pod in the default namespace:
        ```python
        from prefect import flow
        from prefect_kubernetes.credentials import KubernetesCredentials
        from prefect_kubernetes.pods import replace_namespaced_pod
        from kubernetes_asyncio.client.models import V1Pod

        @flow
        def kubernetes_orchestrator():
            v1_pod_metadata = replace_namespaced_pod(
                kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
                pod_name="test-pod",
                new_pod=V1Pod(metadata={"labels": {"foo": "bar"}})
            )
        ```
    """
    async with kubernetes_credentials.get_client("core") as core_v1_client:
        return await core_v1_client.replace_namespaced_pod(
            body=new_pod,
            name=pod_name,
            namespace=namespace,
            **kube_kwargs,
        )