Patch subresources via separate API endpoints

pykube/objects.py

@@ -94,8 +94,11 @@ def api_kwargs(self, **kwargs):
         if obj_list:
             kw["url"] = self.endpoint
         else:
+            subresource = kwargs.pop("subresource", None) or ""
             operation = kwargs.pop("operation", "")
-            kw["url"] = op.normpath(op.join(self.endpoint, self.name, operation))
+            kw["url"] = op.normpath(
+                op.join(self.endpoint, self.name, subresource, operation)
+            )
         params = kwargs.pop("params", None)
         if params is not None:
             query_string = urlencode(params)
@@ -138,25 +141,26 @@ def watch(self):
             .watch()
         )
 
-    def patch(self, strategic_merge_patch):
+    def patch(self, strategic_merge_patch, *, subresource=None):
         """
         Patch the Kubernetes resource by calling the API with a "strategic merge" patch.
         """
         r = self.api.patch(
             **self.api_kwargs(
+                subresource=subresource,
                 headers={"Content-Type": "application/merge-patch+json"},
                 data=json.dumps(strategic_merge_patch),
             )
         )
         self.api.raise_for_status(r)
         self.set_obj(r.json())
 
-    def update(self, is_strategic=True):
+    def update(self, is_strategic=True, *, subresource=None):
         """
         Update the Kubernetes resource by calling the API (patch)
         """
         self.obj = obj_merge(self.obj, self._original_obj, is_strategic)
-        self.patch(self.obj)
+        self.patch(self.obj, subresource=subresource)
 
     def delete(self, propagation_policy: str = None):
         """

tests/test_api.py

@@ -338,3 +338,31 @@ def test_resource_list(api, requests_mock):
         )
         resource_list = api.resource_list("example.org/v1")
         assert resource_list == data2
+
+
+def test_patch_subresource(api, requests_mock):
+    with requests_mock as rsps:
+        rsps.add(
+            responses.GET,
+            "https://localhost:9443/apis/apps/v1/namespaces/default/deployments",
+            json={"items": [{"metadata": {"name": "deploy-1"}}]},
+        )
+
+        deployments = list(Deployment.objects(api))
+        assert len(deployments) == 1
+        deploy = deployments[0]
+        assert deploy.name == "deploy-1"
+        assert deploy.namespace == "default"
+
+        rsps.add(
+            responses.PATCH,
+            "https://localhost:9443/apis/apps/v1/namespaces/default/deployments/deploy-1/status",
+            json={"metadata": {"name": "deploy-1"}, "status": {"field": "field"}},
+        )
+
+        deploy.patch({"status": {"field": "field"}}, subresource="status")
+        assert len(rsps.calls) == 2
+
+        assert json.loads(rsps.calls[-1].request.body) == {
+            "status": {"field": "field"},
+        }