Spaces:
Sleeping
Sleeping
File size: 4,012 Bytes
acd245a |
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 |
"""Admin router module for managing Aider sessions and configurations."""
import os
import shutil
import tempfile
from fastapi import APIRouter, Depends, HTTPException, Security
from fastapi.security import APIKeyHeader
from app.core.config import settings
from app.models.admin import ConfigResponse, ConfigUpdate, SessionInfo, SessionList
router = APIRouter(prefix="/admin", tags=["admin"])
# Security
API_KEY_NAME = "X-API-Key"
api_key_header = APIKeyHeader(name=API_KEY_NAME)
# Define dependency outside of function signature
verify_admin_dep = Security(api_key_header)
async def verify_admin_key(api_key: str = verify_admin_dep) -> str:
"""Verify the admin API key."""
if api_key != settings.ADMIN_API_KEY:
raise HTTPException(status_code=403, detail="Invalid API key")
return api_key
@router.get("/sessions", response_model=SessionList)
async def list_sessions(api_key: str = Depends(verify_admin_key)) -> SessionList:
"""List all active Aider sessions."""
sessions = []
temp_dir = tempfile.gettempdir()
for dir_name in os.listdir(temp_dir):
if dir_name.startswith(settings.TEMP_DIR_PREFIX):
session_path = os.path.join(temp_dir, dir_name)
if os.path.isdir(session_path):
sessions.append(
SessionInfo(
id=dir_name,
path=session_path,
created_at=os.path.getctime(session_path),
size=sum(
os.path.getsize(os.path.join(session_path, f))
for f in os.listdir(session_path)
if os.path.isfile(os.path.join(session_path, f))
),
)
)
return SessionList(sessions=sessions)
@router.delete("/sessions/{session_id}")
async def delete_session(
session_id: str, api_key: str = Depends(verify_admin_key)
) -> dict:
"""Delete a specific Aider session."""
session_path = os.path.join(tempfile.gettempdir(), session_id)
if not os.path.exists(session_path):
raise HTTPException(status_code=404, detail="Session not found")
try:
shutil.rmtree(session_path)
return {"message": "Session deleted successfully"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.delete("/sessions")
async def delete_all_sessions(api_key: str = Depends(verify_admin_key)) -> dict:
"""Delete all Aider sessions."""
temp_dir = tempfile.gettempdir()
deleted_count = 0
for dir_name in os.listdir(temp_dir):
if dir_name.startswith(settings.TEMP_DIR_PREFIX):
session_path = os.path.join(temp_dir, dir_name)
if os.path.isdir(session_path):
try:
shutil.rmtree(session_path)
deleted_count += 1
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
return {"message": f"Deleted {deleted_count} sessions"}
@router.get("/config", response_model=ConfigResponse)
async def get_config(api_key: str = Depends(verify_admin_key)) -> ConfigResponse:
"""Get current configuration settings."""
masked_key = (
settings.OPENAI_API_KEY[:8] + "..." if settings.OPENAI_API_KEY else None
)
return ConfigResponse(
openai_api_key=masked_key,
aider_model=settings.AIDER_MODEL,
temp_dir_prefix=settings.TEMP_DIR_PREFIX,
)
@router.put("/config")
async def update_config(
config: ConfigUpdate, api_key: str = Depends(verify_admin_key)
) -> dict:
"""Update configuration settings."""
if config.openai_api_key:
settings.OPENAI_API_KEY = config.openai_api_key
if config.aider_model:
settings.AIDER_MODEL = config.aider_model
if config.temp_dir_prefix:
settings.TEMP_DIR_PREFIX = config.temp_dir_prefix
return {"message": "Configuration updated successfully"}
|