Patch¶
api.patch(name, patch) applies a partial update to an existing resource. Kubex supports all three patch strategies used by the Kubernetes API.
Patch types¶
| Type | Import | Content-Type |
|---|---|---|
MergePatch |
kubex.core.patch |
application/merge-patch+json |
StrategicMergePatch |
kubex.core.patch |
application/strategic-merge-patch+json |
JsonPatch |
kubex.core.patch |
application/json-patch+json |
MergePatch¶
Merge patch (RFC 7396) replaces the specified keys and removes any key set to null. Keys not present in the patch are left unchanged. Wrap any resource model in MergePatch(...) to apply it:
from kubex.core.patch import MergePatch
from kubex.k8s.v1_35.apps.v1.deployment import Deployment
from kubex.k8s.v1_35.apps.v1.deployment_spec import DeploymentSpec
from kubex.k8s.v1_35.meta.v1.label_selector import LabelSelector
from kubex.k8s.v1_35.core.v1.pod_spec import PodSpec
from kubex.k8s.v1_35.core.v1.pod_template_spec import PodTemplateSpec
from kubex.k8s.v1_35.core.v1.container import Container
from kubex_core.models.metadata import ObjectMetadata
merge_patch = MergePatch(
Deployment(
spec=DeploymentSpec(
replicas=3,
selector=LabelSelector(match_labels={"app": "example"}),
template=PodTemplateSpec(
metadata=ObjectMetadata(labels={"app": "example"}),
spec=PodSpec(containers=[Container(name="nginx", image="nginx:latest")]),
),
)
)
)
patched = await api.patch("example-deploy", merge_patch)
print(patched.spec.replicas) # 3
StrategicMergePatch¶
Strategic merge patch is a Kubernetes extension that understands list-merging semantics (e.g., merging containers by name rather than replacing the whole array). Use it when you want to add or update a list element without specifying the full list:
from kubex.core.patch import StrategicMergePatch
strategic_patch = StrategicMergePatch(
Deployment(
metadata=ObjectMetadata(
annotations={"example.com/patched": "true"},
)
)
)
patched = await api.patch("example-deploy", strategic_patch)
print(patched.metadata.annotations)
Strategic merge patch is not available for custom resources — use MergePatch or JsonPatch there.
JsonPatch¶
JSON Patch (RFC 6902) describes a sequence of operations (add, remove, replace, move, copy, test) using JSON Pointer paths. Build patches incrementally using the fluent API:
from kubex.core.patch import JsonPatch
json_patch = JsonPatch().add("/metadata/labels/version", "v1")
patched = await api.patch("example-deploy", json_patch)
print(patched.metadata.labels)
All six RFC 6902 operations are supported:
patch = (
JsonPatch()
.add("/metadata/labels/env", "staging")
.replace("/spec/replicas", 2)
.remove("/metadata/annotations/example.com~1old-key")
.test("/metadata/name", "example-deploy")
)
Building paths with JsonPointer¶
JsonPointer is accepted anywhere a path string is, and it handles RFC 6901 escaping for you. Build a pointer by chaining the / operator from a base, or from a tuple of unescaped tokens:
from kubex.core.patch import JsonPatch, JsonPointer
# Chained operator — ideal when a fixed prefix is reused
annotation_path = JsonPointer("/metadata") / "annotations" / "example.com/patched"
# annotation_path == "/metadata/annotations/example.com~1patched"
patch = JsonPatch().add(annotation_path, "true")
# Or from raw tokens
label_path = JsonPointer.from_tokens("metadata", "labels", "version")
patch = patch.replace(label_path, "v2")
JSON Pointer escaping
Per RFC 6901, / inside a key is escaped as ~1 and ~ is escaped as ~0. For example, the annotation key example.com/patched becomes the path segment example.com~1patched.
Kubex can do this for you — pass a JsonPointer instead of a raw string:
JsonPointer.from_tokens("metadata", "annotations", "example.com/patched")
or JsonPointer("/metadata") / "annotations" / "example.com/patched". Both
yield /metadata/annotations/example.com~1patched with the slash escaped
automatically.
You can also construct a JsonPatch from a list of operation objects directly:
from kubex.core.patch import JsonPatch
from kubex.core.json_patch import JsonPatchAdd, JsonPatchRemove
patch = JsonPatch([
JsonPatchAdd(path="/metadata/labels/env", value="staging"),
JsonPatchRemove(path="/metadata/labels/old-env"),
])
Patch options¶
All three patch types accept these optional keyword arguments on api.patch():
| Parameter | Description |
|---|---|
dry_run |
Validate without persisting (True or DryRun.ALL) |
field_manager |
Field manager name for server-side apply tracking |
force |
Force apply even when field ownership conflicts (server-side apply only) |
field_validation |
FieldValidation.STRICT (default), WARN, or IGNORE |
patched = await api.patch(
"example-deploy",
merge_patch,
dry_run=True,
field_manager="my-controller",
)
Full example from examples/patch_deployment.py¶
from typing import cast
from kubex.api import Api
from kubex.client import create_client
from kubex.core.patch import JsonPatch, MergePatch, StrategicMergePatch
from kubex.k8s.v1_35.apps.v1.deployment import Deployment
from kubex.k8s.v1_35.apps.v1.deployment_spec import DeploymentSpec
from kubex.k8s.v1_35.core.v1.container import Container
from kubex.k8s.v1_35.core.v1.pod_spec import PodSpec
from kubex.k8s.v1_35.core.v1.pod_template_spec import PodTemplateSpec
from kubex.k8s.v1_35.meta.v1.label_selector import LabelSelector
from kubex_core.models.metadata import ObjectMetadata
async def main() -> None:
client = await create_client()
async with client:
api: Api[Deployment] = Api(Deployment, client=client, namespace="default")
deployment = await api.create(
Deployment(
metadata=ObjectMetadata(name="example-deploy", labels={"app": "example"}),
spec=DeploymentSpec(
replicas=1,
selector=LabelSelector(match_labels={"app": "example"}),
template=PodTemplateSpec(
metadata=ObjectMetadata(labels={"app": "example"}),
spec=PodSpec(containers=[Container(name="nginx", image="nginx:latest")]),
),
),
),
)
name = cast(str, deployment.metadata.name)
try:
# MergePatch — update replicas
merge_patch = MergePatch(
Deployment(
spec=DeploymentSpec(
replicas=3,
selector=LabelSelector(match_labels={"app": "example"}),
template=PodTemplateSpec(
metadata=ObjectMetadata(labels={"app": "example"}),
spec=PodSpec(containers=[Container(name="nginx", image="nginx:latest")]),
),
)
)
)
patched = await api.patch(name, merge_patch)
print(f"After MergePatch: replicas={patched.spec and patched.spec.replicas}")
# StrategicMergePatch — add an annotation
strategic_patch = StrategicMergePatch(
Deployment(metadata=ObjectMetadata(annotations={"example.com/patched": "true"}))
)
patched = await api.patch(name, strategic_patch)
print(f"After StrategicMergePatch: annotations={patched.metadata.annotations}")
# JsonPatch — add a label
json_patch = JsonPatch().add("/metadata/labels/version", "v1")
patched = await api.patch(name, json_patch)
print(f"After JsonPatch: labels={patched.metadata.labels}")
finally:
await api.delete(name)