diff --git a/backend/app/api/v1/dag.py b/backend/app/api/v1/dag.py index 3641ffe..947f4ad 100644 --- a/backend/app/api/v1/dag.py +++ b/backend/app/api/v1/dag.py @@ -1,5 +1,6 @@ """DAG orchestration endpoints.""" -from fastapi import APIRouter, Depends +import json +from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from pydantic import BaseModel from typing import List @@ -7,6 +8,7 @@ from app.database import get_db from app.services.dag_service import DAGService from app.api.deps import get_current_user from app.models.user import User +from app.models.task import Task router = APIRouter(prefix="/dag") @@ -16,6 +18,11 @@ class DAGExecuteRequest(BaseModel): task_ids: List[int] +class DependencyRequest(BaseModel): + task_id: int # downstream task + depends_on_id: int # upstream task (must complete first) + + @router.post("/execute") def execute_dag( req: DAGExecuteRequest, @@ -25,3 +32,126 @@ def execute_dag( """Execute multiple tasks respecting dependencies (DAG).""" results = DAGService.execute_dag(db, user.tenant_id, req.project_id, req.task_ids) return {"success": True, "results": results} + + +@router.get("/graph/{project_id}") +def get_dag_graph( + project_id: int, + user: User = Depends(get_current_user), + db: Session = Depends(get_db), +): + """Get the DAG graph for a project: nodes (tasks) + edges (dependencies).""" + tasks = db.query(Task).filter( + Task.project_id == project_id, Task.tenant_id == user.tenant_id + ).order_by(Task.id).all() + + task_ids = {t.id for t in tasks} + nodes = [] + edges = [] + for t in tasks: + deps = [] + try: + deps = [d for d in json.loads(t.depends_on or "[]") if d in task_ids] + except (json.JSONDecodeError, TypeError): + deps = [] + nodes.append({ + "id": t.id, + "title": t.title, + "status": t.status, + "priority": t.priority, + "task_type": t.task_type, + "worker_id": t.worker_id, + "assignee_id": t.assignee_id, + "requires_review": t.requires_review, + }) + for d in deps: + edges.append({"id": f"e{d}-{t.id}", "source": d, "target": t.id}) + + return {"project_id": project_id, "nodes": nodes, "edges": edges} + + +@router.post("/dependencies", status_code=201) +def add_dependency( + req: DependencyRequest, + user: User = Depends(get_current_user), + db: Session = Depends(get_db), +): + """Add a dependency: task_id depends on depends_on_id. Rejects cycles.""" + if req.task_id == req.depends_on_id: + raise HTTPException(status_code=400, detail="任务不能依赖自身") + + task = db.query(Task).filter( + Task.id == req.task_id, Task.tenant_id == user.tenant_id + ).first() + upstream = db.query(Task).filter( + Task.id == req.depends_on_id, Task.tenant_id == user.tenant_id + ).first() + if not task or not upstream: + raise HTTPException(status_code=404, detail="任务不存在") + if task.project_id != upstream.project_id: + raise HTTPException(status_code=400, detail="只能依赖同一项目内的任务") + + deps = json.loads(task.depends_on or "[]") + if req.depends_on_id in deps: + return {"success": True, "message": "依赖已存在", "depends_on": deps} + + # Cycle detection: would adding task <- upstream create a cycle? + if _creates_cycle(db, user.tenant_id, task.project_id, req.task_id, req.depends_on_id): + raise HTTPException(status_code=400, detail="不能添加该依赖:会形成循环依赖") + + deps.append(req.depends_on_id) + task.depends_on = json.dumps(deps) + db.commit() + return {"success": True, "message": "依赖已添加", "depends_on": deps} + + +@router.delete("/dependencies") +def remove_dependency( + req: DependencyRequest, + user: User = Depends(get_current_user), + db: Session = Depends(get_db), +): + """Remove a dependency edge.""" + task = db.query(Task).filter( + Task.id == req.task_id, Task.tenant_id == user.tenant_id + ).first() + if not task: + raise HTTPException(status_code=404, detail="任务不存在") + + deps = json.loads(task.depends_on or "[]") + if req.depends_on_id not in deps: + raise HTTPException(status_code=404, detail="依赖关系不存在") + + deps.remove(req.depends_on_id) + task.depends_on = json.dumps(deps) + db.commit() + return {"success": True, "message": "依赖已移除", "depends_on": deps} + + +def _creates_cycle(db: Session, tenant_id: int, project_id: int, task_id: int, upstream_id: int) -> bool: + """Check if making task_id depend on upstream_id creates a cycle. + + A cycle exists iff upstream_id (transitively) already depends on task_id. + """ + tasks = db.query(Task).filter( + Task.project_id == project_id, Task.tenant_id == tenant_id + ).all() + dep_map = {} + for t in tasks: + try: + dep_map[t.id] = json.loads(t.depends_on or "[]") + except (json.JSONDecodeError, TypeError): + dep_map[t.id] = [] + + # BFS from upstream_id following its dependency chain; if we reach task_id -> cycle + visited = set() + stack = [upstream_id] + while stack: + cur = stack.pop() + if cur == task_id: + return True + if cur in visited: + continue + visited.add(cur) + stack.extend(dep_map.get(cur, [])) + return False