tasks.py 8.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217
  1. """Task management router for Web-Local system dispatch (P4-M2)."""
  2. from typing import Optional, List, Dict, Any
  3. from fastapi import APIRouter, HTTPException, Depends
  4. from pydantic import BaseModel, Field
  5. from sqlalchemy.orm import Session
  6. from ..database import get_db
  7. from ..services.task_manager import get_task_manager
  8. router = APIRouter(prefix="/api/tasks", tags=["Tasks"])
  9. class CreateTaskRequest(BaseModel):
  10. """Request to create a new simulation task."""
  11. plan_id: Optional[int] = Field(default=None, description="Associated plan ID")
  12. plan_data: Optional[Dict[str, Any]] = Field(default=None, description="Full plan data (auto-loaded from plan_id when omitted)")
  13. parameters: Optional[List[Dict[str, Any]]] = Field(default=None, description="Parameter sets (auto-expanded from plan_id when omitted)")
  14. task_name: Optional[str] = Field(default=None, description="Optional task name")
  15. priority: int = Field(default=5, ge=1, le=10, description="Task priority (1-10)")
  16. class ProgressUpdateRequest(BaseModel):
  17. """Request to update task progress."""
  18. current_point: int = Field(..., ge=0, description="Current point index (0-based)")
  19. total_points: Optional[int] = Field(default=None, description="Total points")
  20. current_params: Optional[Dict[str, Any]] = Field(default=None)
  21. elapsed_time: Optional[float] = Field(default=None)
  22. class ResultsReportRequest(BaseModel):
  23. """Request to report final results."""
  24. results: List[Dict[str, Any]] = Field(..., description="List of simulation results")
  25. metrics: Optional[Dict[str, Any]] = Field(default=None)
  26. logs: Optional[str] = Field(default=None)
  27. duration: Optional[float] = Field(default=None)
  28. status: str = Field(default="completed", description="completed / failed")
  29. @router.post("")
  30. def create_task(request: CreateTaskRequest, db: Session = Depends(get_db)):
  31. """Create a new simulation task.
  32. When only plan_id is provided, plan_data is loaded from the plan and
  33. parameters are expanded from its variables (same logic as the plan's
  34. one-click start), so the UI no longer requires hand-written JSON.
  35. Explicit plan_data / parameters still take precedence (advanced override).
  36. """
  37. plan_data = request.plan_data
  38. parameters = request.parameters
  39. if (not plan_data or not parameters) and request.plan_id is not None:
  40. from ..models.simulation_plan import SimulationPlan
  41. from .plans import _expand_plan_to_parameters
  42. plan = db.query(SimulationPlan).filter(SimulationPlan.id == request.plan_id).first()
  43. if not plan:
  44. raise HTTPException(status_code=404, detail=f"Plan {request.plan_id} not found")
  45. if not plan_data:
  46. plan_data = plan.get_plan_dict()
  47. if not parameters:
  48. parameters = _expand_plan_to_parameters(plan_data)
  49. if not plan_data:
  50. raise HTTPException(status_code=400, detail="plan_data is required (or provide a valid plan_id)")
  51. if not parameters:
  52. raise HTTPException(status_code=400, detail="parameters is empty (plan has no valid variables to simulate)")
  53. try:
  54. manager = get_task_manager()
  55. task = manager.create_task(
  56. plan_id=request.plan_id,
  57. plan_data=plan_data,
  58. parameters=parameters,
  59. task_name=request.task_name,
  60. priority=request.priority,
  61. )
  62. return task
  63. except Exception as e:
  64. raise HTTPException(status_code=500, detail=f"Task creation failed: {str(e)}")
  65. @router.get("")
  66. def list_tasks(
  67. status: Optional[str] = None,
  68. plan_id: Optional[int] = None,
  69. limit: int = 50,
  70. offset: int = 0,
  71. ):
  72. """List tasks with optional filters."""
  73. manager = get_task_manager()
  74. return manager.list_tasks(status=status, plan_id=plan_id, limit=limit, offset=offset)
  75. @router.get("/{task_id}")
  76. def get_task(task_id: str):
  77. """Get task details by ID."""
  78. manager = get_task_manager()
  79. task = manager.get_task(task_id)
  80. if not task:
  81. raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
  82. return task
  83. @router.post("/{task_id}/dispatch")
  84. def dispatch_task(task_id: str):
  85. """Mark task as dispatched (ready for local executor)."""
  86. try:
  87. manager = get_task_manager()
  88. return manager.dispatch_task(task_id)
  89. except ValueError as e:
  90. raise HTTPException(status_code=400, detail=str(e))
  91. except Exception as e:
  92. raise HTTPException(status_code=500, detail=str(e))
  93. @router.post("/{task_id}/progress")
  94. def update_progress(task_id: str, request: ProgressUpdateRequest):
  95. """Update task progress from local executor."""
  96. try:
  97. manager = get_task_manager()
  98. return manager.update_progress(
  99. task_id=task_id,
  100. current_point=request.current_point,
  101. total_points=request.total_points,
  102. current_params=request.current_params,
  103. elapsed_time=request.elapsed_time,
  104. )
  105. except ValueError as e:
  106. raise HTTPException(status_code=404, detail=str(e))
  107. except Exception as e:
  108. raise HTTPException(status_code=500, detail=str(e))
  109. @router.post("/{task_id}/results")
  110. def report_results(task_id: str, request: ResultsReportRequest):
  111. """Report final results from local executor."""
  112. try:
  113. manager = get_task_manager()
  114. return manager.report_results(
  115. task_id=task_id,
  116. results=request.results,
  117. metrics=request.metrics,
  118. logs=request.logs,
  119. duration=request.duration,
  120. status=request.status,
  121. )
  122. except ValueError as e:
  123. raise HTTPException(status_code=404, detail=str(e))
  124. except Exception as e:
  125. raise HTTPException(status_code=500, detail=str(e))
  126. @router.get("/{task_id}/results")
  127. def get_task_results(task_id: str):
  128. """Get task results file content."""
  129. manager = get_task_manager()
  130. results = manager.get_task_results(task_id)
  131. if results is None:
  132. raise HTTPException(status_code=404, detail=f"Results for task {task_id} not found")
  133. return results
  134. @router.post("/{task_id}/cancel")
  135. def cancel_task(task_id: str):
  136. """Cancel a pending or running task."""
  137. try:
  138. manager = get_task_manager()
  139. return manager.cancel_task(task_id)
  140. except ValueError as e:
  141. raise HTTPException(status_code=400, detail=str(e))
  142. except Exception as e:
  143. raise HTTPException(status_code=500, detail=str(e))
  144. @router.delete("/{task_id}", status_code=204)
  145. def delete_task(task_id: str):
  146. """Delete a finished task (terminal status only)."""
  147. try:
  148. manager = get_task_manager()
  149. manager.delete_task(task_id)
  150. return None
  151. except ValueError as e:
  152. raise HTTPException(status_code=400, detail=str(e))
  153. except Exception as e:
  154. raise HTTPException(status_code=500, detail=str(e))
  155. class BatchDeleteTasksRequest(BaseModel):
  156. """Request to delete multiple finished tasks."""
  157. task_ids: List[str] = Field(..., description="Task IDs to delete")
  158. @router.post("/batch-delete")
  159. def batch_delete_tasks(request: BatchDeleteTasksRequest):
  160. """Delete multiple finished tasks; per-item failures do not abort the batch."""
  161. manager = get_task_manager()
  162. deleted, errors = [], []
  163. for task_id in request.task_ids:
  164. try:
  165. manager.delete_task(task_id)
  166. deleted.append(task_id)
  167. except ValueError as e:
  168. errors.append({"task_id": task_id, "error": str(e)})
  169. return {"deleted": deleted, "deleted_count": len(deleted), "errors": errors}
  170. @router.get("/{task_id}/download")
  171. def download_task_file(task_id: str):
  172. """Download task JSON file for local executor."""
  173. manager = get_task_manager()
  174. task = manager.get_task(task_id)
  175. if not task or not task.get("task_file"):
  176. raise HTTPException(status_code=404, detail=f"Task file for {task_id} not found")
  177. import os
  178. if not os.path.exists(task["task_file"]):
  179. raise HTTPException(status_code=404, detail="Task file not found on disk")
  180. from fastapi.responses import FileResponse
  181. return FileResponse(task["task_file"], filename=f"{task_id}_task.json", media_type="application/json")