File size: 3,827 Bytes
b190b45 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 |
#!/usr/bin/env python3
"""
Configuration API Router
========================
API endpoints for configuration management and hot reload
"""
from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import JSONResponse
from typing import Optional, Dict, Any
import logging
from backend.services.config_manager import get_config_manager
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/api/config",
tags=["Configuration"]
)
# Get global config manager instance
config_manager = get_config_manager()
@router.post("/reload")
async def reload_config(config_name: Optional[str] = Query(None, description="Specific config to reload (reloads all if omitted)")) -> JSONResponse:
"""
Manually reload configuration files.
Reloads a specific configuration file or all configuration files.
Args:
config_name: Optional specific config name to reload
Returns:
JSON response with reload status
"""
try:
result = config_manager.manual_reload(config_name)
if result["success"]:
return JSONResponse(
status_code=200,
content={
"success": True,
"message": result["message"],
"data": result
}
)
else:
raise HTTPException(status_code=404, detail=result["message"])
except Exception as e:
logger.error(f"Error reloading config: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
@router.get("/status")
async def get_config_status() -> JSONResponse:
"""
Get configuration status.
Returns the status of all loaded configurations.
Returns:
JSON response with config status
"""
try:
all_configs = config_manager.get_all_configs()
status = {
"loaded_configs": list(all_configs.keys()),
"config_count": len(all_configs),
"configs": {}
}
for config_name, config_data in all_configs.items():
status["configs"][config_name] = {
"version": config_data.get("version", "unknown"),
"last_updated": config_data.get("last_updated", "unknown"),
"keys": list(config_data.keys())
}
return JSONResponse(
status_code=200,
content={
"success": True,
"data": status
}
)
except Exception as e:
logger.error(f"Error getting config status: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
@router.get("/{config_name}")
async def get_config(config_name: str) -> JSONResponse:
"""
Get a specific configuration.
Retrieves the current configuration for a specific config name.
Args:
config_name: Name of the config to retrieve
Returns:
JSON response with configuration data
"""
try:
config = config_manager.get_config(config_name)
if config is None:
raise HTTPException(status_code=404, detail=f"Config '{config_name}' not found")
return JSONResponse(
status_code=200,
content={
"success": True,
"config_name": config_name,
"data": config
}
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Error getting config: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
|