158 lines
5.1 KiB
Python
158 lines
5.1 KiB
Python
"""DAG orchestration endpoints."""
|
|
import json
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from pydantic import BaseModel
|
|
from typing import List
|
|
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")
|
|
|
|
|
|
class DAGExecuteRequest(BaseModel):
|
|
project_id: int
|
|
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,
|
|
user: User = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""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
|