admin.py
3.58 KB
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
"""
Admin API routes for configuration and management.
"""
from fastapi import APIRouter, HTTPException, Request
from ..models import HealthResponse, ErrorResponse
from indexer.mapping_generator import get_tenant_index_name
router = APIRouter(prefix="/admin", tags=["admin"])
@router.get("/health", response_model=HealthResponse)
async def health_check():
"""
Health check endpoint.
Returns service status and Elasticsearch connectivity.
"""
try:
from ..app import get_es_client, get_config
es_client = get_es_client()
config = get_config()
# Check ES connectivity
es_status = "connected" if es_client.ping() else "disconnected"
return HealthResponse(
status="healthy" if es_status == "connected" else "unhealthy",
elasticsearch=es_status
)
except Exception as e:
return HealthResponse(
status="unhealthy",
elasticsearch="error"
)
@router.get("/config")
async def get_configuration():
"""
Get the effective application configuration (sanitized).
"""
try:
from ..app import get_config
return get_config().sanitized_dict()
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/config/meta")
async def get_configuration_meta():
"""Get configuration metadata for observability."""
try:
from ..app import get_config
config = get_config()
return {
"environment": config.runtime.environment,
"config_hash": config.metadata.config_hash,
"loaded_files": list(config.metadata.loaded_files),
"deprecated_keys": list(config.metadata.deprecated_keys),
}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/stats")
async def get_index_stats(http_request: Request):
"""
Get index statistics.
"""
try:
from urllib.parse import parse_qs
from ..app import get_es_client
es_client = get_es_client()
tenant_id = http_request.headers.get("X-Tenant-ID")
if not tenant_id:
query_string = http_request.url.query
if query_string:
params = parse_qs(query_string)
tenant_id = params.get("tenant_id", [None])[0]
if not tenant_id:
raise HTTPException(
status_code=400,
detail="tenant_id is required. Provide it via header 'X-Tenant-ID' or query parameter 'tenant_id'",
)
index_name = get_tenant_index_name(tenant_id)
if not es_client.client.indices.exists(index=index_name):
raise HTTPException(
status_code=404,
detail=f"Tenant index not found: {index_name}",
)
# Get document count
doc_count = es_client.client.count(index=index_name).get("count", 0)
# Get index size (if available)
try:
stats = es_client.client.indices.stats(index=index_name)
size_in_bytes = stats["indices"][index_name]["total"]["store"]["size_in_bytes"]
size_mb = size_in_bytes / (1024 * 1024)
except Exception:
size_mb = None
return {
"tenant_id": str(tenant_id),
"index_name": index_name,
"document_count": doc_count,
"size_mb": round(size_mb, 2) if size_mb else None
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))