"""System settings router.""" from typing import List from fastapi import APIRouter, Depends, Request from sqlalchemy.orm import Session from app.auth import get_current_user, AdminUser from app.database import get_db from app.models.db_models import SystemSetting from app.services import airwallex_service from app.services.audit_log import create_audit_log router = APIRouter(prefix="/api/settings", tags=["settings"]) # Keys that should have their values masked in responses SENSITIVE_KEYS = {"airwallex_api_key", "proxy_password"} @router.get("") def get_settings( db: Session = Depends(get_db), user: AdminUser = Depends(get_current_user), ): """Get all system settings.""" settings = db.query(SystemSetting).all() result = [] for s in settings: result.append({ "key": s.key, "value": s.value, "encrypted": s.encrypted, "updated_at": s.updated_at.isoformat() if s.updated_at else None, }) return result @router.put("") def update_settings( updates: List[dict], request: Request, db: Session = Depends(get_db), user: AdminUser = Depends(get_current_user), ): """Update system settings. Accepts a list of {key, value} objects.""" for item in updates: key = item.get("key") value = item.get("value") if not key: continue # Skip if masked value sent back unchanged if value == "********": continue existing = db.query(SystemSetting).filter(SystemSetting.key == key).first() if existing: existing.value = str(value) existing.encrypted = key in SENSITIVE_KEYS else: db.add(SystemSetting( key=key, value=str(value), encrypted=key in SENSITIVE_KEYS, )) db.commit() create_audit_log( db, action="update_settings", resource_type="settings", operator=user.username, ip_address=request.client.host if request.client else "", details=f"Updated keys: {[item.get('key') for item in updates]}", ) return {"message": "Settings updated"} @router.post("/test-connection") def test_connection( db: Session = Depends(get_db), user: AdminUser = Depends(get_current_user), ): """Test Airwallex API connection with current settings.""" return airwallex_service.test_connection(db) @router.post("/test-proxy") def test_proxy( db: Session = Depends(get_db), user: AdminUser = Depends(get_current_user), ): """Test proxy connectivity and query outbound IP.""" return airwallex_service.test_proxy(db)