Files
orion/app/api/v1/admin/settings.py

218 lines
5.8 KiB
Python

# app/api/v1/admin/settings.py
"""
Platform settings management endpoints.
Provides endpoints for:
- Viewing all platform settings
- Creating/updating settings
- Managing configuration by category
"""
import logging
from typing import Optional
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from app.api.deps import get_current_admin_user
from app.core.database import get_db
from app.services.admin_settings_service import admin_settings_service
from app.services.admin_audit_service import admin_audit_service
from models.schema.admin import (
AdminSettingCreate,
AdminSettingResponse,
AdminSettingUpdate,
AdminSettingListResponse
)
from models.database.user import User
router = APIRouter(prefix="/settings")
logger = logging.getLogger(__name__)
@router.get("", response_model=AdminSettingListResponse)
def get_all_settings(
category: Optional[str] = Query(None, description="Filter by category"),
is_public: Optional[bool] = Query(None, description="Filter by public flag"),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""
Get all platform settings.
Can be filtered by category (system, security, marketplace, notifications)
and by public flag (settings that can be exposed to frontend).
"""
settings = admin_settings_service.get_all_settings(db, category, is_public)
return AdminSettingListResponse(
settings=settings,
total=len(settings),
category=category
)
@router.get("/categories")
def get_setting_categories(
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""Get list of all setting categories."""
# This could be enhanced to return counts per category
return {
"categories": [
"system",
"security",
"marketplace",
"notifications",
"integrations",
"payments"
]
}
@router.get("/{key}", response_model=AdminSettingResponse)
def get_setting(
key: str,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""Get specific setting by key."""
setting = admin_settings_service.get_setting_by_key(db, key)
if not setting:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail=f"Setting '{key}' not found")
return AdminSettingResponse.model_validate(setting)
@router.post("", response_model=AdminSettingResponse)
def create_setting(
setting_data: AdminSettingCreate,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""
Create new platform setting.
Setting keys should be lowercase with underscores (e.g., max_vendors_allowed).
"""
result = admin_settings_service.create_setting(
db=db,
setting_data=setting_data,
admin_user_id=current_admin.id
)
# Log action
admin_audit_service.log_action(
db=db,
admin_user_id=current_admin.id,
action="create_setting",
target_type="setting",
target_id=setting_data.key,
details={"category": setting_data.category, "value_type": setting_data.value_type}
)
return result
@router.put("/{key}", response_model=AdminSettingResponse)
def update_setting(
key: str,
update_data: AdminSettingUpdate,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""Update existing setting value."""
old_value = admin_settings_service.get_setting_value(db, key)
result = admin_settings_service.update_setting(
db=db,
key=key,
update_data=update_data,
admin_user_id=current_admin.id
)
# Log action
admin_audit_service.log_action(
db=db,
admin_user_id=current_admin.id,
action="update_setting",
target_type="setting",
target_id=key,
details={"old_value": str(old_value), "new_value": update_data.value}
)
return result
@router.post("/upsert", response_model=AdminSettingResponse)
def upsert_setting(
setting_data: AdminSettingCreate,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""
Create or update setting (upsert).
If setting exists, updates its value. If not, creates new setting.
"""
result = admin_settings_service.upsert_setting(
db=db,
setting_data=setting_data,
admin_user_id=current_admin.id
)
# Log action
admin_audit_service.log_action(
db=db,
admin_user_id=current_admin.id,
action="upsert_setting",
target_type="setting",
target_id=setting_data.key,
details={"category": setting_data.category}
)
return result
@router.delete("/{key}")
def delete_setting(
key: str,
confirm: bool = Query(False, description="Must be true to confirm deletion"),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_user),
):
"""
Delete platform setting.
Requires confirmation parameter.
WARNING: Deleting settings may affect platform functionality.
"""
from fastapi import HTTPException
if not confirm:
raise HTTPException(
status_code=400,
detail="Deletion requires confirmation parameter: confirm=true"
)
message = admin_settings_service.delete_setting(
db=db,
key=key,
admin_user_id=current_admin.id
)
# Log action
admin_audit_service.log_action(
db=db,
admin_user_id=current_admin.id,
action="delete_setting",
target_type="setting",
target_id=key,
details={}
)
return {"message": message}