add knowledgebase-file from form
(
self, knowledgebase_id: str, files: list[str]
)
| 313 | raise |
| 314 | |
| 315 | def add_knowledgebase_file_form( |
| 316 | self, knowledgebase_id: str, files: list[str] |
| 317 | ) -> MemOSAddKnowledgebaseFileResponse | None: |
| 318 | """ |
| 319 | add knowledgebase-file from form |
| 320 | """ |
| 321 | # Validate required parameters |
| 322 | self._validate_required_params(knowledgebase_id=knowledgebase_id, files=files) |
| 323 | |
| 324 | def build_file_form_param(file_path): |
| 325 | """ |
| 326 | form-Automatically generate the structure required for the `files` parameter in requests based on the local file path |
| 327 | """ |
| 328 | if not os.path.isfile(file_path): |
| 329 | logger.warning(f"File {file_path} does not exist") |
| 330 | return None |
| 331 | filename = os.path.basename(file_path) |
| 332 | |
| 333 | mime_type, _ = mimetypes.guess_type(file_path) |
| 334 | if mime_type is None: |
| 335 | mime_type = "application/octet-stream" |
| 336 | return ("file", (filename, open(file_path, "rb"), mime_type)) |
| 337 | |
| 338 | url = f"{self.base_url}/add/knowledgebase-file" |
| 339 | payload = { |
| 340 | "knowledgebase_id": knowledgebase_id, |
| 341 | } |
| 342 | headers = { |
| 343 | "Authorization": f"Token {self.api_key}", |
| 344 | } |
| 345 | for retry in range(MAX_RETRY_COUNT): |
| 346 | try: |
| 347 | response = requests.post( |
| 348 | url, |
| 349 | params=payload, |
| 350 | headers=headers, |
| 351 | timeout=30, |
| 352 | files=[build_file_form_param(file_path) for file_path in files], |
| 353 | ) |
| 354 | response.raise_for_status() |
| 355 | response_data = response.json() |
| 356 | print(response_data) |
| 357 | |
| 358 | return MemOSAddKnowledgebaseFileResponse(**response_data) |
| 359 | except Exception as e: |
| 360 | logger.error(f"Failed to add knowledgebase-file form (retry {retry + 1}/3): {e}") |
| 361 | if retry == MAX_RETRY_COUNT - 1: |
| 362 | raise |
| 363 | |
| 364 | def delete_knowledgebase_file( |
| 365 | self, file_ids: list[str] |
nothing calls this directly
no test coverage detected