feat(api): support new worker update labels api

Author: BobAnkh

src/pynetmito/__init__.py

@@ -50,6 +50,7 @@
     UpdateTaskLabelsReq,
     ChangeTaskReq,
     ReplaceWorkerTagsReq,
+    ReplaceWorkerLabelsReq,
     UpdateGroupWorkerRoleReq,
     RemoveGroupWorkerRoleReq,
     UpdateUserGroupRoleReq,
@@ -110,6 +111,7 @@
     "UpdateTaskLabelsReq",
     "ChangeTaskReq",
     "ReplaceWorkerTagsReq",
+    "ReplaceWorkerLabelsReq",
     "UpdateGroupWorkerRoleReq",
     "RemoveGroupWorkerRoleReq",
     "UpdateUserGroupRoleReq",

src/pynetmito/client.py

@@ -39,6 +39,7 @@
     UpdateTaskLabelsReq,
     ChangeTaskReq,
     ReplaceWorkerTagsReq,
+    ReplaceWorkerLabelsReq,
     UpdateGroupWorkerRoleReq,
     RemoveGroupWorkerRoleReq,
     UpdateUserGroupRoleReq,
@@ -663,6 +664,18 @@ def replace_worker_tags(self, uuid: UUID4, req: ReplaceWorkerTagsReq):
                 f"Failed to replace worker tags for {str(uuid)}, status code: {resp.status_code}, error: {resp.text}"
             )
 
+    def replace_worker_labels(self, uuid: UUID4, req: ReplaceWorkerLabelsReq):
+        url = self._get_url(f"workers/{str(uuid)}/labels")
+        headers = {"Authorization": f"Bearer {self.credential}"}
+        resp = self.http_client.put(url, headers=headers, json=req.to_dict())
+        if resp.status_code == 200:
+            return
+        else:
+            self.logger.error(resp.text)
+            raise Exception(
+                f"Failed to replace worker labels for {str(uuid)}, status code: {resp.status_code}, error: {resp.text}"
+            )
+
     def update_group_worker_roles(self, uuid: UUID4, req: UpdateGroupWorkerRoleReq):
         url = self._get_url(f"workers/{str(uuid)}/groups")
         headers = {"Authorization": f"Bearer {self.credential}"}

src/pynetmito/schemas.py

@@ -852,6 +852,19 @@ def deserialize_tags(cls, tags: list[str]):
         return set(tags)
 
 
+class ReplaceWorkerLabelsReq(BaseAPIModel):
+    labels: Set[str]
+
+    @field_serializer("labels")
+    def serialize_labels(self, labels: Set[str]):
+        return list(labels)
+
+    @field_validator("labels", mode="before")
+    @classmethod
+    def deserialize_labels(cls, labels: list[str]):
+        return set(labels)
+
+
 class UpdateGroupWorkerRoleReq(BaseAPIModel):
     relations: Dict[str, GroupWorkerRole]