6240d68d09
- Add require_engineer_or_admin to POST/PUT/DELETE in target_lists.py (blocks viewers from write ops) - Add require_engineer_or_admin to POST/PATCH in maintenance_schedules.py (blocks viewers from write ops) - Add team ownership guard in batch_launch_sessions after active/published checks (Fix 2) - Wrap scheduler.remove_job in try/except for SchedulerNotRunningError and JobLookupError (Fix 3) - Recompute next_run_at when is_active flips to True, capturing was_active before update (Fix 4) - Add optional batch_id and target_label fields to Session type; remove unsafe cast in MaintenanceFlowDetailPage.tsx (Fix 5) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
"""Target lists CRUD endpoints."""
|
|
from typing import Annotated
|
|
from uuid import UUID
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.deps import get_current_active_user, get_db, require_engineer_or_admin
|
|
from app.models.target_list import TargetList
|
|
from app.models.user import User
|
|
from app.schemas.target_list import TargetListCreate, TargetListUpdate, TargetListResponse
|
|
|
|
router = APIRouter(prefix="/target-lists", tags=["target-lists"])
|
|
|
|
|
|
@router.get("/", response_model=list[TargetListResponse])
|
|
async def list_target_lists(
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
):
|
|
"""List all target lists for the current user's team."""
|
|
if not current_user.team_id:
|
|
return []
|
|
result = await db.execute(
|
|
select(TargetList)
|
|
.where(TargetList.team_id == current_user.team_id)
|
|
.order_by(TargetList.name)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.post("/", response_model=TargetListResponse, status_code=201)
|
|
async def create_target_list(
|
|
data: TargetListCreate,
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
_: None = Depends(require_engineer_or_admin),
|
|
):
|
|
"""Create a new target list for the current team."""
|
|
if not current_user.team_id:
|
|
raise HTTPException(status_code=400, detail="User must belong to a team")
|
|
target_list = TargetList(
|
|
team_id=current_user.team_id,
|
|
created_by=current_user.id,
|
|
name=data.name,
|
|
description=data.description,
|
|
targets=[t.model_dump() for t in data.targets],
|
|
)
|
|
db.add(target_list)
|
|
await db.commit()
|
|
await db.refresh(target_list)
|
|
return target_list
|
|
|
|
|
|
@router.get("/{list_id}", response_model=TargetListResponse)
|
|
async def get_target_list(
|
|
list_id: UUID,
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
):
|
|
result = await db.execute(
|
|
select(TargetList).where(
|
|
TargetList.id == list_id,
|
|
TargetList.team_id == current_user.team_id,
|
|
)
|
|
)
|
|
target_list = result.scalar_one_or_none()
|
|
if not target_list:
|
|
raise HTTPException(status_code=404, detail="Target list not found")
|
|
return target_list
|
|
|
|
|
|
@router.put("/{list_id}", response_model=TargetListResponse)
|
|
async def update_target_list(
|
|
list_id: UUID,
|
|
data: TargetListUpdate,
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
_: None = Depends(require_engineer_or_admin),
|
|
):
|
|
result = await db.execute(
|
|
select(TargetList).where(
|
|
TargetList.id == list_id,
|
|
TargetList.team_id == current_user.team_id,
|
|
)
|
|
)
|
|
target_list = result.scalar_one_or_none()
|
|
if not target_list:
|
|
raise HTTPException(status_code=404, detail="Target list not found")
|
|
update_fields = data.model_fields_set
|
|
if "name" in update_fields and data.name is not None:
|
|
target_list.name = data.name
|
|
if "description" in update_fields:
|
|
target_list.description = data.description # allow setting to None
|
|
if "targets" in update_fields and data.targets is not None:
|
|
target_list.targets = [t.model_dump() for t in data.targets]
|
|
await db.commit()
|
|
await db.refresh(target_list)
|
|
return target_list
|
|
|
|
|
|
@router.delete("/{list_id}", status_code=204)
|
|
async def delete_target_list(
|
|
list_id: UUID,
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
_: None = Depends(require_engineer_or_admin),
|
|
):
|
|
result = await db.execute(
|
|
select(TargetList).where(
|
|
TargetList.id == list_id,
|
|
TargetList.team_id == current_user.team_id,
|
|
)
|
|
)
|
|
target_list = result.scalar_one_or_none()
|
|
if not target_list:
|
|
raise HTTPException(status_code=404, detail="Target list not found")
|
|
await db.delete(target_list)
|
|
await db.commit()
|