"""Milestones API router.""" from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from typing import List from app.core.config import get_db from app.api.deps import get_current_user_or_apikey from app.api.rbac import check_project_role from app.models import models from app.models.milestone import Milestone from app.schemas import schemas router = APIRouter(prefix="/projects/{project_id}/milestones", tags=["Milestones"]) @router.get("", response_model=List[schemas.MilestoneResponse]) def list_milestones(project_id: int, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user_or_apikey)): """List all milestones for a project.""" check_project_role(db, current_user.id, project_id, min_role="viewer") milestones = db.query(Milestone).filter(models.Milestone.project_id == project_id).all() return milestones @router.post("", response_model=schemas.MilestoneResponse, status_code=status.HTTP_201_CREATED) def create_milestone(project_id: int, milestone: schemas.MilestoneCreate, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user_or_apikey)): """Create a new milestone for a project.""" check_project_role(db, current_user.id, project_id, min_role="mgr") db_milestone = Milestone(project_id=project_id, **milestone.model_dump()) db.add(db_milestone) db.commit() db.refresh(db_milestone) return db_milestone @router.get("/{milestone_id}", response_model=schemas.MilestoneResponse) def get_milestone(project_id: int, milestone_id: int, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user_or_apikey)): """Get a milestone by ID.""" check_project_role(db, current_user.id, project_id, min_role="viewer") milestone = db.query(Milestone).filter(models.Milestone.id == milestone_id, models.Milestone.project_id == project_id).first() if not milestone: raise HTTPException(status_code=404, detail="Milestone not found") return milestone @router.patch("/{milestone_id}", response_model=schemas.MilestoneResponse) def update_milestone(project_id: int, milestone_id: int, milestone: schemas.MilestoneUpdate, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user_or_apikey)): """Update a milestone.""" check_project_role(db, current_user.id, project_id, min_role="mgr") db_milestone = db.query(Milestone).filter(models.Milestone.id == milestone_id, models.Milestone.project_id == project_id).first() if not db_milestone: raise HTTPException(status_code=404, detail="Milestone not found") for key, value in milestone.model_dump(exclude_unset=True).items(): setattr(db_milestone, key, value) db.commit() db.refresh(db_milestone) return db_milestone @router.delete("/{milestone_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_milestone(project_id: int, milestone_id: int, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user_or_apikey)): """Delete a milestone.""" check_project_role(db, current_user.id, project_id, min_role="admin") db_milestone = db.query(Milestone).filter(models.Milestone.id == milestone_id, models.Milestone.project_id == project_id).first() if not db_milestone: raise HTTPException(status_code=404, detail="Milestone not found") db.delete(db_milestone) db.commit() return None