from fastapi import Depends, FastAPI, HTTPException, status from app.config import get_settings from app.embedding import embed_text, embed_texts from app.schemas import ( BatchEmbedRequest, BatchEmbedResponse, EmbedRequest, EmbedResponse, HealthResponse, ) from app.security import require_api_key app = FastAPI(title="ACCESS SBERT Embedding Service") def _validate_text(text: str, max_length: int) -> str: trimmed = text.strip() if not trimmed: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Text is required", ) if len(trimmed) > max_length: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Text too long", ) return trimmed @app.get("/health", response_model=HealthResponse) def health() -> dict[str, int | str]: settings = get_settings() return { "status": "ok", "model": settings.model_name, "dimension": settings.dimension, } @app.post("/embed", response_model=EmbedResponse) def embed( request: EmbedRequest, _: None = Depends(require_api_key), ) -> dict[str, int | list[float] | str]: settings = get_settings() text = _validate_text(request.text, settings.max_text_length) vector = embed_text(text) return { "embedding": vector, "dimension": len(vector), "model": settings.model_name, } @app.post("/embed/batch", response_model=BatchEmbedResponse) def embed_batch( request: BatchEmbedRequest, _: None = Depends(require_api_key), ) -> dict[str, int | list[list[float]] | str]: settings = get_settings() if len(request.texts) > settings.max_batch_size: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Too many texts", ) texts = [ _validate_text(text, settings.max_text_length) for text in request.texts ] vectors = embed_texts(texts) return { "embeddings": vectors, "dimension": len(vectors[0]) if vectors else 0, "model": settings.model_name, }