Vito Sansevero
2737a4b1ef
Automated formatting pass across 30 files to establish consistent code style enforced by CI. No logic changes.
208 lines
7.2 KiB
Python
208 lines
7.2 KiB
Python
"""Logging API routes for PromptManager."""
|
|
|
|
import os
|
|
|
|
from aiohttp import web
|
|
|
|
|
|
class LoggingRoutesMixin:
|
|
"""Mixin providing logging-related API endpoints."""
|
|
|
|
def _register_logging_routes(self, routes):
|
|
@routes.get("/prompt_manager/logs")
|
|
async def get_logs_route(request):
|
|
return await self.get_logs(request)
|
|
|
|
@routes.get("/prompt_manager/logs/files")
|
|
async def get_log_files_route(request):
|
|
return await self.get_log_files(request)
|
|
|
|
@routes.get("/prompt_manager/logs/download/{filename}")
|
|
async def download_log_route(request):
|
|
return await self.download_log_file(request)
|
|
|
|
@routes.post("/prompt_manager/logs/truncate")
|
|
async def truncate_logs_route(request):
|
|
return await self.truncate_logs(request)
|
|
|
|
@routes.get("/prompt_manager/logs/config")
|
|
async def get_log_config_route(request):
|
|
return await self.get_log_config(request)
|
|
|
|
@routes.post("/prompt_manager/logs/config")
|
|
async def update_log_config_route(request):
|
|
return await self.update_log_config(request)
|
|
|
|
@routes.get("/prompt_manager/logs/stats")
|
|
async def get_log_stats_route(request):
|
|
return await self.get_log_stats(request)
|
|
|
|
def _get_logger_manager(self):
|
|
"""Get the logger manager instance."""
|
|
try:
|
|
from ...utils.logging_config import get_logger_manager
|
|
except ImportError:
|
|
import sys
|
|
|
|
current_dir = os.path.dirname(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
)
|
|
sys.path.insert(0, current_dir)
|
|
from utils.logging_config import get_logger_manager
|
|
return get_logger_manager()
|
|
|
|
async def get_logs(self, request):
|
|
"""Get recent log entries."""
|
|
try:
|
|
logger_manager = self._get_logger_manager()
|
|
|
|
limit = int(request.query.get("limit", 100))
|
|
level = request.query.get("level", None)
|
|
|
|
if limit > 1000:
|
|
limit = 1000
|
|
elif limit < 1:
|
|
limit = 1
|
|
|
|
logs = logger_manager.get_recent_logs(limit=limit, level=level)
|
|
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"logs": logs,
|
|
"count": len(logs),
|
|
"level_filter": level,
|
|
"limit": limit,
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Get logs error: {e}")
|
|
return web.json_response(
|
|
{"success": False, "error": str(e), "logs": []}, status=500
|
|
)
|
|
|
|
async def get_log_files(self, request):
|
|
"""Get information about available log files."""
|
|
try:
|
|
logger_manager = self._get_logger_manager()
|
|
log_files = logger_manager.get_log_files()
|
|
|
|
return web.json_response(
|
|
{"success": True, "files": log_files, "count": len(log_files)}
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Get log files error: {e}")
|
|
return web.json_response(
|
|
{"success": False, "error": str(e), "files": []}, status=500
|
|
)
|
|
|
|
async def download_log_file(self, request):
|
|
"""Download a specific log file."""
|
|
try:
|
|
filename = request.match_info["filename"]
|
|
logger_manager = self._get_logger_manager()
|
|
|
|
if not filename or ".." in filename or "/" in filename or "\\" in filename:
|
|
return web.json_response(
|
|
{"success": False, "error": "Invalid filename"}, status=400
|
|
)
|
|
|
|
log_file_path = logger_manager.log_dir / filename
|
|
|
|
if not log_file_path.exists():
|
|
return web.json_response(
|
|
{"success": False, "error": "Log file not found"}, status=404
|
|
)
|
|
|
|
with open(log_file_path, "rb") as f:
|
|
file_content = f.read()
|
|
|
|
return web.Response(
|
|
body=file_content,
|
|
content_type="text/plain",
|
|
headers={
|
|
"Content-Disposition": f'attachment; filename="{filename}"',
|
|
"Content-Length": str(len(file_content)),
|
|
},
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Download log file error: {e}")
|
|
return web.json_response({"success": False, "error": str(e)}, status=500)
|
|
|
|
async def truncate_logs(self, request):
|
|
"""Truncate all log files."""
|
|
try:
|
|
logger_manager = self._get_logger_manager()
|
|
results = logger_manager.truncate_logs()
|
|
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"message": f"Truncated {len(results['truncated'])} log files",
|
|
"results": results,
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Truncate logs error: {e}")
|
|
return web.json_response({"success": False, "error": str(e)}, status=500)
|
|
|
|
async def get_log_config(self, request):
|
|
"""Get current logging configuration."""
|
|
try:
|
|
logger_manager = self._get_logger_manager()
|
|
config = logger_manager.get_config()
|
|
|
|
return web.json_response({"success": True, "config": config})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Get log config error: {e}")
|
|
return web.json_response({"success": False, "error": str(e)}, status=500)
|
|
|
|
async def update_log_config(self, request):
|
|
"""Update logging configuration."""
|
|
try:
|
|
data = await request.json()
|
|
logger_manager = self._get_logger_manager()
|
|
|
|
if "level" in data:
|
|
valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
|
|
if data["level"].upper() not in valid_levels:
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Invalid log level. Must be one of: {valid_levels}",
|
|
},
|
|
status=400,
|
|
)
|
|
data["level"] = data["level"].upper()
|
|
|
|
logger_manager.update_config(data)
|
|
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"message": "Logging configuration updated",
|
|
"config": logger_manager.get_config(),
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Update log config error: {e}")
|
|
return web.json_response({"success": False, "error": str(e)}, status=500)
|
|
|
|
async def get_log_stats(self, request):
|
|
"""Get logging statistics."""
|
|
try:
|
|
logger_manager = self._get_logger_manager()
|
|
stats = logger_manager.get_log_stats()
|
|
|
|
return web.json_response({"success": True, "stats": stats})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Get log stats error: {e}")
|
|
return web.json_response({"success": False, "error": str(e)}, status=500)
|