(task: Dict[str, Any])
| 457 | |
| 458 | @staticmethod |
| 459 | def _domain_task_snapshot(task: Dict[str, Any]) -> Dict[str, Any]: |
| 460 | return { |
| 461 | "id": task.get("id"), |
| 462 | "domain": task.get("domain"), |
| 463 | "task_type": task.get("task_type"), |
| 464 | "status": task.get("status"), |
| 465 | "message": task.get("message"), |
| 466 | "created_at": task.get("created_at"), |
| 467 | "started_at": task.get("started_at"), |
| 468 | "finished_at": task.get("finished_at"), |
| 469 | "cancel_requested": bool(task.get("cancel_requested")), |
| 470 | "pause_requested": bool(task.get("pause_requested")), |
| 471 | "paused": bool(task.get("paused")), |
| 472 | "retry_count": int(task.get("retry_count") or 0), |
| 473 | "max_retries": int(task.get("max_retries") or 0), |
| 474 | "payload": dict(task.get("payload") or {}), |
| 475 | "progress": dict(task.get("progress") or {}), |
| 476 | "result": task.get("result"), |
| 477 | "error": task.get("error"), |
| 478 | "details": list(task.get("details") or []), |
| 479 | } |
| 480 | |
| 481 | def set_domain_quota(self, domain: str, quota: int) -> int: |
| 482 | domain_key = str(domain or "").strip().lower() |
no test coverage detected