66 lines
1.6 KiB
Python
66 lines
1.6 KiB
Python
"""API endpoints for AIOS Search Gateway."""
|
|
|
|
from typing import Any
|
|
|
|
try:
|
|
from fastapi import APIRouter
|
|
except Exception:
|
|
APIRouter = None
|
|
|
|
from .search_gateway import SearchGateway, SearchRequest
|
|
from .search_gateway_dashboard import SearchGatewayDashboard
|
|
|
|
|
|
gateway = SearchGateway()
|
|
dashboard = SearchGatewayDashboard()
|
|
|
|
|
|
def search_gateway_request(payload: dict[str, Any]) -> dict[str, Any]:
|
|
request = SearchRequest(**payload)
|
|
return gateway.search(request).to_dict()
|
|
|
|
|
|
def get_search_gateway_status() -> dict[str, Any]:
|
|
return {
|
|
"module": "search-gateway",
|
|
"status": "ready",
|
|
"providers_configured": len(gateway.providers),
|
|
"policy_guard": "enabled",
|
|
"profile_customization": "enabled",
|
|
"fallback_required_when_provider_unavailable": True,
|
|
}
|
|
|
|
|
|
def get_search_gateway_overview() -> dict[str, Any]:
|
|
return dashboard.overview()
|
|
|
|
|
|
def build_router():
|
|
if APIRouter is None:
|
|
return None
|
|
router = APIRouter(prefix="/admin/ai-evolution/search", tags=["search-gateway"])
|
|
|
|
@router.get("/status", operation_id="getSearchGatewayStatus")
|
|
def status():
|
|
return get_search_gateway_status()
|
|
|
|
@router.get("/overview", operation_id="getSearchGatewayOverview")
|
|
def overview():
|
|
return get_search_gateway_overview()
|
|
|
|
@router.post("/query", operation_id="runSearchGatewayQuery")
|
|
def query(payload: dict[str, Any]):
|
|
return search_gateway_request(payload)
|
|
|
|
return router
|
|
|
|
|
|
router = build_router()
|
|
|
|
__all__ = [
|
|
"router",
|
|
"search_gateway_request",
|
|
"get_search_gateway_status",
|
|
"get_search_gateway_overview",
|
|
]
|