search.py (1747B)
1 """Search routes — proxy search queries to Hermes API.""" 2 3 from __future__ import annotations 4 5 import logging 6 from typing import Any 7 8 from fastapi import APIRouter 9 from pydantic import BaseModel 10 11 from app.config import get_config 12 from app.hermes_client import HermesClient 13 14 logger = logging.getLogger(__name__) 15 router = APIRouter(prefix="/api/search", tags=["search"]) 16 17 18 class SearchResultGroup(BaseModel): 19 session_id: str 20 session_name: str | None = None 21 messages: list[dict[str, Any]] 22 23 24 class SearchResponse(BaseModel): 25 results: list[SearchResultGroup] 26 27 28 def get_hermes_client() -> HermesClient: # pragma: no mutate: block 29 cfg = get_config() 30 return HermesClient(base_url=cfg.hermes_base_url, api_key=cfg.hermes_api_key) 31 32 33 @router.get("") 34 async def search_messages(q: str = "") -> SearchResponse: # pragma: no mutate: block 35 if not q.strip(): 36 return SearchResponse(results=[]) 37 client = get_hermes_client() 38 try: 39 data = await client.search_messages(q) 40 except Exception as exc: 41 # The Hermes API may not have a /api/search endpoint. 42 # Log the error and return empty results gracefully. 43 logger.warning("Search failed (endpoint may not exist): %s", exc) 44 return SearchResponse(results=[]) 45 if data is None: 46 return SearchResponse(results=[]) 47 raw = data.get("results", data) if isinstance(data, dict) else data 48 if not isinstance(raw, list): 49 return SearchResponse(results=[]) 50 results = [ 51 SearchResultGroup( 52 session_id=g.get("session_id", ""), 53 session_name=g.get("session_name"), 54 messages=g.get("messages", []), 55 ) 56 for g in raw 57 ] 58 return SearchResponse(results=results)