get task status
(self, task_id: str)
| 418 | raise |
| 419 | |
| 420 | def get_task_status(self, task_id: str) -> MemOSGetTaskStatusResponse | None: |
| 421 | """ |
| 422 | get task status |
| 423 | """ |
| 424 | # Validate required parameters |
| 425 | self._validate_required_params(task_id=task_id) |
| 426 | |
| 427 | url = f"{self.base_url}/get/status" |
| 428 | payload = { |
| 429 | "task_id": task_id, |
| 430 | } |
| 431 | |
| 432 | for retry in range(MAX_RETRY_COUNT): |
| 433 | try: |
| 434 | response = requests.post( |
| 435 | url, data=json.dumps(payload), headers=self.headers, timeout=30 |
| 436 | ) |
| 437 | response.raise_for_status() |
| 438 | response_data = response.json() |
| 439 | |
| 440 | return MemOSGetTaskStatusResponse(**response_data) |
| 441 | except Exception as e: |
| 442 | logger.error(f"Failed to get task status (retry {retry + 1}/3): {e}") |
| 443 | if retry == MAX_RETRY_COUNT - 1: |
| 444 | raise |
| 445 | |
| 446 | def add_feedback( |
| 447 | self, |
no test coverage detected