feat(fase4): add GlobalVariable API routes
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>
This commit is contained in:
95
services/api-gateway/app/routers/global_variables.py
Normal file
95
services/api-gateway/app/routers/global_variables.py
Normal file
@@ -0,0 +1,95 @@
|
||||
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}
|
||||
Reference in New Issue
Block a user