(self, domain: str, task_id: str, **fields)
| 517 | return self._domain_task_snapshot(task) |
| 518 | |
| 519 | def update_domain_task(self, domain: str, task_id: str, **fields) -> Optional[Dict[str, Any]]: |
| 520 | with _domain_lock: |
| 521 | task_type = fields.pop("task_type", None) |
| 522 | task = self._ensure_domain_task_locked( |
| 523 | domain=domain, |
| 524 | task_id=task_id, |
| 525 | task_type=str(task_type) if task_type is not None else None, |
| 526 | ) |
| 527 | progress = fields.pop("progress", None) |
| 528 | details = fields.pop("details", None) |
| 529 | if progress is not None: |
| 530 | task.setdefault("progress", {}).update(dict(progress or {})) |
| 531 | if details is not None: |
| 532 | task["details"] = list(details or []) |
| 533 | task.update(fields) |
| 534 | if task.get("status") in {"completed", "failed", "cancelled"}: |
| 535 | _domain_running.get(str(domain).strip().lower(), set()).discard(str(task_id)) |
| 536 | return self._domain_task_snapshot(task) |
| 537 | |
| 538 | def append_domain_task_detail(self, domain: str, task_id: str, detail: Dict[str, Any], max_items: int = 500) -> None: |
| 539 | with _domain_lock: |
no test coverage detected