diff --git a/packages/leann-core/src/leann/server.py b/packages/leann-core/src/leann/server.py index e58bfe65..70d4e1c5 100644 --- a/packages/leann-core/src/leann/server.py +++ b/packages/leann-core/src/leann/server.py @@ -25,7 +25,7 @@ def _ensure_fastapi(): """Lazy import FastAPI and Pydantic, with a clear error if missing.""" try: - from fastapi import FastAPI, HTTPException # type: ignore + from fastapi import Body, FastAPI, HTTPException # type: ignore from pydantic import BaseModel # type: ignore except ImportError as e: # pragma: no cover - dependency error path raise RuntimeError( @@ -34,7 +34,7 @@ def _ensure_fastapi(): " uv pip install 'fastapi>=0.115' 'pydantic>=2' 'uvicorn[standard]'\n" ) from e - return FastAPI, HTTPException, BaseModel + return FastAPI, HTTPException, BaseModel, Body def _resolve_index_path(index_name: str) -> str: @@ -92,7 +92,7 @@ def create_app(): - POST /indexes/{name}/search -> semantic search """ - FastAPI, HTTPException, BaseModel = _ensure_fastapi() + FastAPI, HTTPException, BaseModel, Body = _ensure_fastapi() class SearchRequest(BaseModel): query: str @@ -132,10 +132,11 @@ async def list_indexes() -> list[dict[str, Any]]: return _list_current_project_indexes() @app.post("/indexes/{index_name}/search", response_model=list[SearchResultModel]) - async def search_index(index_name: str, body: SearchRequest): + async def search_index(index_name: str, body: dict[str, Any] = Body(...)): """ Run semantic search against an existing LEANN index. """ + parsed = SearchRequest(**body) try: index_path = _resolve_index_path(index_name) except FileNotFoundError as e: @@ -143,14 +144,14 @@ async def search_index(index_name: str, body: SearchRequest): searcher = LeannSearcher(index_path=index_path) results = searcher.search( - query=body.query, - top_k=body.top_k, - complexity=body.complexity, - beam_width=body.beam_width, - prune_ratio=body.prune_ratio, - recompute_embeddings=body.recompute_embeddings, - pruning_strategy=body.pruning_strategy, # type: ignore[arg-type] - use_grep=body.use_grep, + query=parsed.query, + top_k=parsed.top_k, + complexity=parsed.complexity, + beam_width=parsed.beam_width, + prune_ratio=parsed.prune_ratio, + recompute_embeddings=parsed.recompute_embeddings, + pruning_strategy=parsed.pruning_strategy, # type: ignore[arg-type] + use_grep=parsed.use_grep, ) # Normalize into JSON-serializable structures diff --git a/tests/test_http_server_api.py b/tests/test_http_server_api.py new file mode 100644 index 00000000..52a35e2d --- /dev/null +++ b/tests/test_http_server_api.py @@ -0,0 +1,61 @@ +import pytest + +pytest.importorskip("fastapi") + +from fastapi.testclient import TestClient + +from leann import server as leann_server + + +class _DummySearchResult: + def __init__(self): + self.id = "doc-1" + self.score = 0.99 + self.text = "dummy result" + self.metadata = {"source": "test"} + + +class _DummySearcher: + def __init__(self, index_path: str): + self.index_path = index_path + + def search(self, **kwargs): + return [_DummySearchResult()] + + +def test_openapi_schema_loads(): + app = leann_server.create_app() + client = TestClient(app) + + response = client.get("/openapi.json") + + assert response.status_code == 200 + schema = response.json() + assert "/indexes/{index_name}/search" in schema["paths"] + + +def test_search_endpoint_accepts_json_body(monkeypatch): + monkeypatch.setattr(leann_server, "_resolve_index_path", lambda _: "/tmp/fake.leann") + monkeypatch.setattr(leann_server, "LeannSearcher", _DummySearcher) + + app = leann_server.create_app() + client = TestClient(app) + + response = client.post( + "/indexes/demo/search", + json={ + "query": "travel insurance", + "top_k": 2, + "complexity": 32, + "beam_width": 1, + "prune_ratio": 0.0, + "recompute_embeddings": True, + "pruning_strategy": "global", + "use_grep": False, + }, + ) + + assert response.status_code == 200 + data = response.json() + assert len(data) == 1 + assert data[0]["id"] == "doc-1"