68 lines
3.4 KiB
Python
68 lines
3.4 KiB
Python
"""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.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(models.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 = models.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(models.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(models.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(models.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
|