Added in_use and priority_id field for channels urls. Added priorities table. Setup sql alchemy migration. Generate first migration.
All checks were successful
AWS Deploy on Push / build (push) Successful in 2m4s

This commit is contained in:
2025-05-26 21:24:41 -05:00
parent 76dc8908de
commit 21cc99eff6
15 changed files with 547 additions and 35 deletions

97
app/routers/priorities.py Normal file
View File

@@ -0,0 +1,97 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from sqlalchemy import select, delete
from typing import List
from app.models.db import Priority
from app.models.schemas import PriorityCreate, PriorityResponse
from app.utils.database import get_db
from app.auth.dependencies import get_current_user, require_roles
from app.models.auth import CognitoUser
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