Function main

examples_asyncio/patch.py:23–103

()

async def main():

    await config.load_kube_config()

    async with ApiClient() as api:

        v1 = client.CoreV1Api(api)

        print(f"Recreate {SERVICE_NAME}...")
        try:
            await v1.read_namespaced_service(SERVICE_NAME, SERVICE_NS)
            await v1.delete_namespaced_service(SERVICE_NAME, SERVICE_NS)
        except client.exceptions.ApiException as ex:
            if ex.status == 404:
                pass

        await v1.create_namespaced_service(SERVICE_NS, SERVICE_SPEC)

        print("Patch using JSON patch - replace port-80 with port-1000")
        patch = [
            {
                "op": "replace",
                "path": "/spec/ports/0",
                "value": {
                    "name": "port-1000",
                    "protocol": "TCP",
                    "port": 1000,
                    "targetPort": 1000,
                },
            }
        ]
        await v1.patch_namespaced_service(
            SERVICE_NAME,
            SERVICE_NS,
            patch,
            # _content_type='application/json-patch+json'  # (optional, default if patch is a list)
        )

        print(
            "Patch using strategic merge patch - add port-2000, service will have two ports: port-1000 and port-2000"
        )
        patch = {
            "spec": {
                "ports": [
                    {
                        "name": "port-2000",
                        "protocol": "TCP",
                        "port": 2000,
                        "targetPort": 2000,
                    }
                ]
            }
        }
        await v1.patch_namespaced_service(
            SERVICE_NAME,
            SERVICE_NS,
            patch,
            # _content_type='application/strategic-merge-patch+json'  # (optional, default if patch is a dict)