feat(api): support batch operation for artifacts, attachments and tasks

Author: BobAnkh

src/pynetmito/client.py

@@ -62,6 +62,16 @@
     AttachmentsDownloadByFilterReq,
     AttachmentsDownloadByKeysReq,
     AttachmentsDownloadListResp,
+    ArtifactsDeleteByFilterReq,
+    ArtifactsDeleteByFilterResp,
+    ArtifactsDeleteByUuidsReq,
+    ArtifactsDeleteByUuidsResp,
+    AttachmentsDeleteByFilterReq,
+    AttachmentsDeleteByFilterResp,
+    AttachmentsDeleteByKeysReq,
+    AttachmentsDeleteByKeysResp,
+    TasksSubmitReq,
+    TasksSubmitResp,
 )
 
 
@@ -946,3 +956,81 @@ def admin_shutdown_coordinator(self, req: ShutdownReq):
             raise Exception(
                 f"Failed to shutdown coordinator, status code: {resp.status_code}, error: {resp.text}"
             )
+
+    def batch_delete_artifacts_by_filter(
+        self, req: ArtifactsDeleteByFilterReq
+    ) -> ArtifactsDeleteByFilterResp:
+        """Batch delete artifacts by filter criteria."""
+        url = self._get_url("tasks/delete/artifacts")
+        headers = {"Authorization": f"Bearer {self.credential}"}
+        resp = self.http_client.post(url, headers=headers, json=req.to_dict())
+        if resp.status_code == 200:
+            r = ArtifactsDeleteByFilterResp.model_validate(resp.json())
+            return r
+        else:
+            self.logger.error(resp.text)
+            raise Exception(
+                f"Failed to batch delete artifacts by filter, status code: {resp.status_code}, error: {resp.text}"
+            )
+
+    def batch_delete_artifacts_by_list(
+        self, req: ArtifactsDeleteByUuidsReq
+    ) -> ArtifactsDeleteByUuidsResp:
+        """Batch delete artifacts by task UUIDs."""
+        url = self._get_url("tasks/delete/artifacts/list")
+        headers = {"Authorization": f"Bearer {self.credential}"}
+        resp = self.http_client.post(url, headers=headers, json=req.to_dict())
+        if resp.status_code == 200:
+            r = ArtifactsDeleteByUuidsResp.model_validate(resp.json())
+            return r
+        else:
+            self.logger.error(resp.text)
+            raise Exception(
+                f"Failed to batch delete artifacts by list, status code: {resp.status_code}, error: {resp.text}"
+            )
+
+    def batch_delete_attachments_by_filter(
+        self, group_name: str, req: AttachmentsDeleteByFilterReq
+    ) -> AttachmentsDeleteByFilterResp:
+        """Batch delete attachments by filter criteria."""
+        url = self._get_url(f"groups/{group_name}/delete/attachments")
+        headers = {"Authorization": f"Bearer {self.credential}"}
+        resp = self.http_client.post(url, headers=headers, json=req.to_dict())
+        if resp.status_code == 200:
+            r = AttachmentsDeleteByFilterResp.model_validate(resp.json())
+            return r
+        else:
+            self.logger.error(resp.text)
+            raise Exception(
+                f"Failed to batch delete attachments by filter for group {group_name}, status code: {resp.status_code}, error: {resp.text}"
+            )
+
+    def batch_delete_attachments_by_list(
+        self, group_name: str, req: AttachmentsDeleteByKeysReq
+    ) -> AttachmentsDeleteByKeysResp:
+        """Batch delete attachments by keys."""
+        url = self._get_url(f"groups/{group_name}/delete/attachments/list")
+        headers = {"Authorization": f"Bearer {self.credential}"}
+        resp = self.http_client.post(url, headers=headers, json=req.to_dict())
+        if resp.status_code == 200:
+            r = AttachmentsDeleteByKeysResp.model_validate(resp.json())
+            return r
+        else:
+            self.logger.error(resp.text)
+            raise Exception(
+                f"Failed to batch delete attachments by list for group {group_name}, status code: {resp.status_code}, error: {resp.text}"
+            )
+
+    def batch_submit_tasks(self, req: TasksSubmitReq) -> TasksSubmitResp:
+        """Batch submit tasks."""
+        url = self._get_url("tasks/submit")
+        headers = {"Authorization": f"Bearer {self.credential}"}
+        resp = self.http_client.post(url, headers=headers, json=req.to_dict())
+        if resp.status_code == 200:
+            r = TasksSubmitResp.model_validate(resp.json())
+            return r
+        else:
+            self.logger.error(resp.text)
+            raise Exception(
+                f"Failed to batch submit tasks, status code: {resp.status_code}, error: {resp.text}"
+            )

src/pynetmito/schemas.py

@@ -1168,3 +1168,152 @@ class AttachmentsDownloadListResp(BaseAPIModel):
 
     downloads: list[AttachmentDownloadItem]
     group_name: str
+
+
+class ArtifactsDeleteByFilterReq(BaseAPIModel):
+    """Request to batch delete artifacts by filter criteria."""
+
+    model_config = ConfigDict(
+        extra="allow",
+        validate_assignment=True,
+        use_enum_values=True,
+    )
+    creator_usernames: Optional[Set[str]] = Field(default=None)
+    group_name: Optional[str] = Field(default=None)
+    tags: Optional[Set[str]] = Field(default=None)
+    labels: Optional[Set[str]] = Field(default=None)
+    states: Optional[Set[TaskState]] = Field(default=None)
+    exit_status: Optional[str] = Field(default=None)
+    priority: Optional[str] = Field(default=None)
+    content_type: ArtifactContentType
+
+    @field_serializer("creator_usernames")
+    def serialize_creator_usernames(self, creator_usernames: Optional[Set[str]]):
+        return list(creator_usernames) if creator_usernames else None
+
+    @field_validator("creator_usernames", mode="before")
+    @classmethod
+    def deserialize_creator_usernames(cls, creator_usernames: Optional[list[str]]):
+        return set(creator_usernames) if creator_usernames else None
+
+    @field_serializer("tags")
+    def serialize_tags(self, tags: Optional[Set[str]]):
+        return list(tags) if tags else None
+
+    @field_validator("tags", mode="before")
+    @classmethod
+    def deserialize_tags(cls, tags: Optional[list[str]]):
+        return set(tags) if tags else None
+
+    @field_serializer("labels")
+    def serialize_labels(self, labels: Optional[Set[str]]):
+        return list(labels) if labels else None
+
+    @field_validator("labels", mode="before")
+    @classmethod
+    def deserialize_labels(cls, labels: Optional[list[str]]):
+        return set(labels) if labels else None
+
+    @field_serializer("states")
+    def serialize_states(self, states: Optional[Set[TaskState]]):
+        return list(states) if states else None
+
+    @field_validator("states", mode="before")
+    @classmethod
+    def deserialize_states(cls, states: Optional[list[TaskState]]):
+        return set(states) if states else None
+
+
+class ArtifactsDeleteByFilterResp(BaseAPIModel):
+    """Response for batch artifact deletion by filter."""
+
+    deleted_count: NonNegativeInt
+
+
+class ArtifactsDeleteByUuidsReq(BaseAPIModel):
+    """Request to batch delete artifacts by task UUIDs."""
+
+    uuids: list[UUID4]
+    content_type: ArtifactContentType
+
+
+class ArtifactsDeleteByUuidsResp(BaseAPIModel):
+    """Response for batch artifact deletion by UUIDs."""
+
+    deleted_count: NonNegativeInt
+    failed_uuids: list[UUID4]
+
+
+class AttachmentsDeleteByFilterReq(BaseAPIModel):
+    """Request to batch delete attachments by filter criteria."""
+
+    key: Optional[str] = Field(default=None)
+    limit: Optional[NonNegativeInt] = Field(default=None)
+    offset: Optional[NonNegativeInt] = Field(default=None)
+
+
+class AttachmentsDeleteByFilterResp(BaseAPIModel):
+    """Response for batch attachment deletion by filter."""
+
+    deleted_count: NonNegativeInt
+    group_name: str
+
+
+class AttachmentsDeleteByKeysReq(BaseAPIModel):
+    """Request to batch delete attachments by keys."""
+
+    keys: list[str]
+
+
+class AttachmentsDeleteByKeysResp(BaseAPIModel):
+    """Response for batch attachment deletion by keys."""
+
+    deleted_count: NonNegativeInt
+    failed_keys: list[str]
+    group_name: str
+
+
+class TasksSubmitReq(BaseAPIModel):
+    """Request to batch submit tasks."""
+
+    tasks: list[SubmitTaskReq]
+
+
+class ErrorMsg(BaseAPIModel):
+    """Error message wrapper."""
+
+    msg: str
+
+
+class TasksSubmitResp(BaseAPIModel):
+    """Response for batch task submission."""
+
+    results: list[Union[SubmitTaskResp, ErrorMsg]]
+
+    @model_validator(mode="before")
+    @classmethod
+    def deserialize_results(cls, data: Any):
+        if not isinstance(data, dict) or "results" not in data:
+            return data
+
+        results = data["results"]
+        parsed_results = []
+
+        for item in results:
+            if isinstance(item, dict):
+                # Handle Result enum format from Rust
+                if "Ok" in item:
+                    parsed_results.append(SubmitTaskResp(**item["Ok"]))
+                elif "Err" in item:
+                    parsed_results.append(ErrorMsg(**item["Err"]))
+                else:
+                    # Direct object
+                    if "msg" in item:
+                        parsed_results.append(ErrorMsg(**item))
+                    else:
+                        parsed_results.append(SubmitTaskResp(**item))
+            else:
+                parsed_results.append(item)
+
+        data["results"] = parsed_results
+        return data