62d4c37c1d
Add schemas and router for global variables with CRUD operations and admin-only access controls for create/update/delete. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
96 lines
3.0 KiB
Python
96 lines
3.0 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from typing import List
|
|
from uuid import UUID
|
|
from app.core.database import get_db
|
|
from app.core.security import get_current_user
|
|
from app.models.user import User, UserRole
|
|
from app.models.global_variable import GlobalVariable
|
|
from app.schemas.global_variable import (
|
|
GlobalVariableCreate, GlobalVariableUpdate, GlobalVariableResponse
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/global-variables", tags=["global-variables"])
|
|
|
|
|
|
def require_admin(current_user: User = Depends(get_current_user)):
|
|
if current_user.role != UserRole.ADMIN:
|
|
raise HTTPException(status_code=403, detail="Admin required")
|
|
return current_user
|
|
|
|
|
|
@router.get("", response_model=List[GlobalVariableResponse])
|
|
def list_global_variables(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
variables = db.query(GlobalVariable).all()
|
|
for var in variables:
|
|
if var.is_secret:
|
|
var.value = "********"
|
|
return variables
|
|
|
|
|
|
@router.post("", response_model=GlobalVariableResponse)
|
|
def create_global_variable(
|
|
request: GlobalVariableCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_admin),
|
|
):
|
|
existing = db.query(GlobalVariable).filter(GlobalVariable.key == request.key).first()
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="Variable key already exists")
|
|
|
|
variable = GlobalVariable(**request.model_dump())
|
|
db.add(variable)
|
|
db.commit()
|
|
db.refresh(variable)
|
|
return variable
|
|
|
|
|
|
@router.put("/{variable_id}", response_model=GlobalVariableResponse)
|
|
def update_global_variable(
|
|
variable_id: UUID,
|
|
request: GlobalVariableUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_admin),
|
|
):
|
|
variable = db.query(GlobalVariable).filter(GlobalVariable.id == variable_id).first()
|
|
if not variable:
|
|
raise HTTPException(status_code=404, detail="Variable not found")
|
|
|
|
for key, value in request.model_dump(exclude_unset=True).items():
|
|
setattr(variable, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(variable)
|
|
return variable
|
|
|
|
|
|
@router.delete("/{variable_id}")
|
|
def delete_global_variable(
|
|
variable_id: UUID,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_admin),
|
|
):
|
|
variable = db.query(GlobalVariable).filter(GlobalVariable.id == variable_id).first()
|
|
if not variable:
|
|
raise HTTPException(status_code=404, detail="Variable not found")
|
|
|
|
db.delete(variable)
|
|
db.commit()
|
|
return {"success": True}
|
|
|
|
|
|
@router.get("/by-key/{key}")
|
|
def get_variable_by_key(
|
|
key: str,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
variable = db.query(GlobalVariable).filter(GlobalVariable.key == key).first()
|
|
if not variable:
|
|
raise HTTPException(status_code=404, detail="Variable not found")
|
|
|
|
return {"key": variable.key, "value": variable.value if not variable.is_secret else None}
|