Skip to content

prefect_kubernetes.deployments

Module for interacting with Kubernetes deployments from Prefect flows.

Classes

Functions

create_namespaced_deployment async

Create a Kubernetes deployment in a given namespace.

Parameters:

Name Type Description Default
kubernetes_credentials KubernetesCredentials

KubernetesCredentials block for creating authenticated Kubernetes API clients.

 required
new_deployment V1Deployment

A Kubernetes V1Deployment specification.

 required
namespace Optional[str]

The Kubernetes namespace to create this deployment in.

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

Optional extra keyword arguments to pass to the Kubernetes API.

 {}

Returns:

Type Description
V1Deployment

A Kubernetes V1Deployment object.
Example

Create a deployment in the default namespace: 

from prefect import flow
from prefect_kubernetes.credentials import KubernetesCredentials
from prefect_kubernetes.deployments import create_namespaced_deployment
from kubernetes.client.models import V1Deployment

@flow
def kubernetes_orchestrator():
    v1_deployment_metadata = create_namespaced_deployment(
        kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
        new_deployment=V1Deployment(metadata={"name": "test-deployment"}),
    )

Source code in prefect_kubernetes/deployments.py 
11
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_deployment(
    kubernetes_credentials: KubernetesCredentials,
    new_deployment: V1Deployment,
    namespace: Optional[str] = "default",
    **kube_kwargs: Dict[str, Any],
) -> V1Deployment:
    """Create a Kubernetes deployment in a given namespace.

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

    Returns:
        A Kubernetes `V1Deployment` object.

    Example:
        Create a deployment in the default namespace:
        ```python
        from prefect import flow
        from prefect_kubernetes.credentials import KubernetesCredentials
        from prefect_kubernetes.deployments import create_namespaced_deployment
        from kubernetes.client.models import V1Deployment

        @flow
        def kubernetes_orchestrator():
            v1_deployment_metadata = create_namespaced_deployment(
                kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
                new_deployment=V1Deployment(metadata={"name": "test-deployment"}),
            )
        ```
    """
    with kubernetes_credentials.get_client("apps") as apps_v1_client:
        return await run_sync_in_worker_thread(
            apps_v1_client.create_namespaced_deployment,
            namespace=namespace,
            body=new_deployment,
            **kube_kwargs,
        )

delete_namespaced_deployment async

Delete a Kubernetes deployment in a given namespace.

Parameters:

Name Type Description Default
kubernetes_credentials KubernetesCredentials

KubernetesCredentials block for creating authenticated Kubernetes API clients.

 required
deployment_name str

The name of the deployment to delete.

 required
delete_options Optional[V1DeleteOptions]

A Kubernetes V1DeleteOptions object.

 None
namespace Optional[str]

The Kubernetes namespace to delete this deployment from.

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

Optional extra keyword arguments to pass to the Kubernetes API.

 {}

Returns:

Type Description
V1Deployment

A Kubernetes V1Deployment object.
Example

Delete a deployment in the default namespace: 

from prefect import flow
from prefect_kubernetes.credentials import KubernetesCredentials
from prefect_kubernetes.deployments import delete_namespaced_deployment
from kubernetes.client.models import V1DeleteOptions

@flow
def kubernetes_orchestrator():
    v1_deployment_metadata = delete_namespaced_deployment(
        kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
        deployment_name="test-deployment",
        delete_options=V1DeleteOptions(grace_period_seconds=0),
    )

Source code in prefect_kubernetes/deployments.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
100
@task
async def delete_namespaced_deployment(
    kubernetes_credentials: KubernetesCredentials,
    deployment_name: str,
    delete_options: Optional[V1DeleteOptions] = None,
    namespace: Optional[str] = "default",
    **kube_kwargs: Dict[str, Any],
) -> V1Deployment:
    """Delete a Kubernetes deployment in a given namespace.

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

    Returns:
        A Kubernetes `V1Deployment` object.

    Example:
        Delete a deployment in the default namespace:
        ```python
        from prefect import flow
        from prefect_kubernetes.credentials import KubernetesCredentials
        from prefect_kubernetes.deployments import delete_namespaced_deployment
        from kubernetes.client.models import V1DeleteOptions

        @flow
        def kubernetes_orchestrator():
            v1_deployment_metadata = delete_namespaced_deployment(
                kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
                deployment_name="test-deployment",
                delete_options=V1DeleteOptions(grace_period_seconds=0),
            )
        ```
    """
    with kubernetes_credentials.get_client("apps") as apps_v1_client:
        return await run_sync_in_worker_thread(
            apps_v1_client.delete_namespaced_deployment,
            deployment_name,
            body=delete_options,
            namespace=namespace,
            **kube_kwargs,
        )

list_namespaced_deployment async

List all deployments 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 deployments from.

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

Optional extra keyword arguments to pass to the Kubernetes API.

 {}

Returns:

Type Description
V1DeploymentList

A Kubernetes V1DeploymentList object.
Example

List all deployments in the default namespace: 

from prefect import flow
from prefect_kubernetes.credentials import KubernetesCredentials
from prefect_kubernetes.deployments import list_namespaced_deployment

@flow
def kubernetes_orchestrator():
    v1_deployment_list = list_namespaced_deployment(
        kubernetes_credentials=KubernetesCredentials.load("k8s-creds")
    )

Source code in prefect_kubernetes/deployments.py 
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
137
138
139
@task
async def list_namespaced_deployment(
    kubernetes_credentials: KubernetesCredentials,
    namespace: Optional[str] = "default",
    **kube_kwargs: Dict[str, Any],
) -> V1DeploymentList:
    """List all deployments in a given namespace.

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

    Returns:
        A Kubernetes `V1DeploymentList` object.

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

        @flow
        def kubernetes_orchestrator():
            v1_deployment_list = list_namespaced_deployment(
                kubernetes_credentials=KubernetesCredentials.load("k8s-creds")
            )
        ```
    """
    with kubernetes_credentials.get_client("apps") as apps_v1_client:
        return await run_sync_in_worker_thread(
            apps_v1_client.list_namespaced_deployment,
            namespace=namespace,
            **kube_kwargs,
        )

patch_namespaced_deployment async

Patch a Kubernetes deployment in a given namespace.

Parameters:

Name Type Description Default
kubernetes_credentials KubernetesCredentials

KubernetesCredentials block for creating authenticated Kubernetes API clients.

 required
deployment_name str

The name of the deployment to patch.

 required
deployment_updates V1Deployment

A Kubernetes V1Deployment object.

 required
namespace Optional[str]

The Kubernetes namespace to patch this deployment in.

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

Optional extra keyword arguments to pass to the Kubernetes API.

 {}

Returns:

Type Description
V1Deployment

A Kubernetes V1Deployment object.
Example

Patch a deployment in the default namespace: 

from prefect import flow
from prefect_kubernetes.credentials import KubernetesCredentials
from prefect_kubernetes.deployments import patch_namespaced_deployment
from kubernetes.client.models import V1Deployment

@flow
def kubernetes_orchestrator():
    v1_deployment_metadata = patch_namespaced_deployment(
        kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
        deployment_name="test-deployment",
        deployment_updates=V1Deployment(metadata={"labels": {"foo": "bar"}}),
    )

Source code in prefect_kubernetes/deployments.py 
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
184
185
186
187
@task
async def patch_namespaced_deployment(
    kubernetes_credentials: KubernetesCredentials,
    deployment_name: str,
    deployment_updates: V1Deployment,
    namespace: Optional[str] = "default",
    **kube_kwargs: Dict[str, Any],
) -> V1Deployment:
    """Patch a Kubernetes deployment in a given namespace.

    Args:
        kubernetes_credentials: `KubernetesCredentials` block for creating
            authenticated Kubernetes API clients.
        deployment_name: The name of the deployment to patch.
        deployment_updates: A Kubernetes `V1Deployment` object.
        namespace: The Kubernetes namespace to patch this deployment in.
        **kube_kwargs: Optional extra keyword arguments to pass to the Kubernetes API.

    Returns:
        A Kubernetes `V1Deployment` object.

    Example:
        Patch a deployment in the default namespace:
        ```python
        from prefect import flow
        from prefect_kubernetes.credentials import KubernetesCredentials
        from prefect_kubernetes.deployments import patch_namespaced_deployment
        from kubernetes.client.models import V1Deployment

        @flow
        def kubernetes_orchestrator():
            v1_deployment_metadata = patch_namespaced_deployment(
                kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
                deployment_name="test-deployment",
                deployment_updates=V1Deployment(metadata={"labels": {"foo": "bar"}}),
            )
        ```
    """
    with kubernetes_credentials.get_client("apps") as apps_v1_client:
        return await run_sync_in_worker_thread(
            apps_v1_client.patch_namespaced_deployment,
            name=deployment_name,
            namespace=namespace,
            body=deployment_updates,
            **kube_kwargs,
        )

read_namespaced_deployment async

Read information on a Kubernetes deployment in a given namespace.

Parameters:

Name Type Description Default
kubernetes_credentials KubernetesCredentials

KubernetesCredentials block for creating authenticated Kubernetes API clients.

 required
deployment_name str

The name of the deployment to read.

 required
namespace Optional[str]

The Kubernetes namespace to read this deployment from.

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

Optional extra keyword arguments to pass to the Kubernetes API.

 {}

Returns:

Type Description
V1Deployment

A Kubernetes V1Deployment object.
Example

Read a deployment in the default namespace: 

from prefect import flow
from prefect_kubernetes.credentials import KubernetesCredentials

@flow
def kubernetes_orchestrator():
    v1_deployment_metadata = read_namespaced_deployment(
        kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
        deployment_name="test-deployment"
    )

Source code in prefect_kubernetes/deployments.py 
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
225
226
227
228
229
@task
async def read_namespaced_deployment(
    kubernetes_credentials: KubernetesCredentials,
    deployment_name: str,
    namespace: Optional[str] = "default",
    **kube_kwargs: Dict[str, Any],
) -> V1Deployment:
    """Read information on a Kubernetes deployment in a given namespace.

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

    Returns:
        A Kubernetes `V1Deployment` object.

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

        @flow
        def kubernetes_orchestrator():
            v1_deployment_metadata = read_namespaced_deployment(
                kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
                deployment_name="test-deployment"
            )
        ```
    """
    with kubernetes_credentials.get_client("apps") as apps_v1_client:
        return await run_sync_in_worker_thread(
            apps_v1_client.read_namespaced_deployment,
            name=deployment_name,
            namespace=namespace,
            **kube_kwargs,
        )

replace_namespaced_deployment async

Replace a Kubernetes deployment in a given namespace.

Parameters:

Name Type Description Default
kubernetes_credentials KubernetesCredentials

KubernetesCredentials block for creating authenticated Kubernetes API clients.

 required
deployment_name str

The name of the deployment to replace.

 required
new_deployment V1Deployment

A Kubernetes V1Deployment object.

 required
namespace Optional[str]

The Kubernetes namespace to replace this deployment in.

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

Optional extra keyword arguments to pass to the Kubernetes API.

 {}

Returns:

Type Description
V1Deployment

A Kubernetes V1Deployment object.
Example

Replace a deployment in the default namespace: 

from prefect import flow
from prefect_kubernetes.credentials import KubernetesCredentials
from prefect_kubernetes.deployments import replace_namespaced_deployment
from kubernetes.client.models import V1Deployment

@flow
def kubernetes_orchestrator():
    v1_deployment_metadata = replace_namespaced_deployment(
        kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
        deployment_name="test-deployment",
        new_deployment=V1Deployment(metadata={"labels": {"foo": "bar"}})
    )

Source code in prefect_kubernetes/deployments.py 
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
@task
async def replace_namespaced_deployment(
    kubernetes_credentials: KubernetesCredentials,
    deployment_name: str,
    new_deployment: V1Deployment,
    namespace: Optional[str] = "default",
    **kube_kwargs: Dict[str, Any],
) -> V1Deployment:
    """Replace a Kubernetes deployment in a given namespace.

    Args:
        kubernetes_credentials: `KubernetesCredentials` block for creating
            authenticated Kubernetes API clients.
        deployment_name: The name of the deployment to replace.
        new_deployment: A Kubernetes `V1Deployment` object.
        namespace: The Kubernetes namespace to replace this deployment in.
        **kube_kwargs: Optional extra keyword arguments to pass to the Kubernetes API.

    Returns:
        A Kubernetes `V1Deployment` object.

    Example:
        Replace a deployment in the default namespace:
        ```python
        from prefect import flow
        from prefect_kubernetes.credentials import KubernetesCredentials
        from prefect_kubernetes.deployments import replace_namespaced_deployment
        from kubernetes.client.models import V1Deployment

        @flow
        def kubernetes_orchestrator():
            v1_deployment_metadata = replace_namespaced_deployment(
                kubernetes_credentials=KubernetesCredentials.load("k8s-creds"),
                deployment_name="test-deployment",
                new_deployment=V1Deployment(metadata={"labels": {"foo": "bar"}})
            )
        ```
    """
    with kubernetes_credentials.get_client("apps") as apps_v1_client:
        return await run_sync_in_worker_thread(
            apps_v1_client.replace_namespaced_deployment,
            body=new_deployment,
            name=deployment_name,
            namespace=namespace,
            **kube_kwargs,
        )