93 lines
2.7 KiB
Python
93 lines
2.7 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth.dependencies import get_current_user, require_roles
|
|
from app.models.auth import CognitoUser
|
|
from app.models.db import Priority
|
|
from app.models.schemas import PriorityCreate, PriorityResponse
|
|
from app.utils.database import get_db
|
|
|
|
router = APIRouter(prefix="/priorities", tags=["priorities"])
|
|
|
|
|
|
@router.post("/", response_model=PriorityResponse, status_code=status.HTTP_201_CREATED)
|
|
@require_roles("admin")
|
|
def create_priority(
|
|
priority: PriorityCreate,
|
|
db: Session = Depends(get_db),
|
|
user: CognitoUser = Depends(get_current_user),
|
|
):
|
|
"""Create a new priority"""
|
|
# Check if priority with this ID already exists
|
|
existing = db.get(Priority, priority.id)
|
|
if existing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=f"Priority with ID {priority.id} already exists",
|
|
)
|
|
|
|
db_priority = Priority(**priority.model_dump())
|
|
db.add(db_priority)
|
|
db.commit()
|
|
db.refresh(db_priority)
|
|
return db_priority
|
|
|
|
|
|
@router.get("/", response_model=list[PriorityResponse])
|
|
@require_roles("admin")
|
|
def list_priorities(
|
|
db: Session = Depends(get_db), user: CognitoUser = Depends(get_current_user)
|
|
):
|
|
"""List all priorities"""
|
|
return db.query(Priority).all()
|
|
|
|
|
|
@router.get("/{priority_id}", response_model=PriorityResponse)
|
|
@require_roles("admin")
|
|
def get_priority(
|
|
priority_id: int,
|
|
db: Session = Depends(get_db),
|
|
user: CognitoUser = Depends(get_current_user),
|
|
):
|
|
"""Get a priority by id"""
|
|
priority = db.get(Priority, priority_id)
|
|
if not priority:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Priority not found"
|
|
)
|
|
return priority
|
|
|
|
|
|
@router.delete("/{priority_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
@require_roles("admin")
|
|
def delete_priority(
|
|
priority_id: int,
|
|
db: Session = Depends(get_db),
|
|
user: CognitoUser = Depends(get_current_user),
|
|
):
|
|
"""Delete a priority (if not in use)"""
|
|
from app.models.db import ChannelURL
|
|
|
|
# Check if priority exists
|
|
priority = db.get(Priority, priority_id)
|
|
if not priority:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Priority not found"
|
|
)
|
|
|
|
# Check if priority is in use
|
|
in_use = db.scalar(
|
|
select(ChannelURL).where(ChannelURL.priority_id == priority_id).limit(1)
|
|
)
|
|
|
|
if in_use:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Cannot delete priority that is in use by channel URLs",
|
|
)
|
|
|
|
db.execute(delete(Priority).where(Priority.id == priority_id))
|
|
db.commit()
|
|
return None
|