Intégration FastAPI

Un service FastAPI complet exposant la recherche Whoosh-NG via HTTP.

1. Installation

pip install "whoosh-ng[api]" fastapi uvicorn

2. Créer l'index

# setup_index.py
import json
from whoosh import index
from whoosh.fields import Schema, TEXT, ID

schema = Schema(
    id=ID(stored=True, unique=True),
    title=TEXT(stored=True),
    content=TEXT,
)

ix = index.create_in("docs_index", schema)

with ix.writer() as w:
    for doc in json.load(open("documents.json")):
        w.add_document(
            id=doc["id"],
            title=doc["title"],
            content=doc["content"],
        )
    w.commit()

3. Service REST

# main.py
from fastapi import FastAPI, Query
from typing import Optional
from whoosh import index
from whoosh.qparser import QueryParser
from whoosh_fastapi import create_app

ix = index.open_dir("docs_index")

# Option A: Utiliser l'aide
app = create_app(ix, prefix="/api/v1")

# Option B: endpoints manuels
# app = FastAPI(title="Document Search API", version="1.0.0")
#
# @app.get("/api/v1/health")
# async def health():
#     return {"status": "ok"}
#
# @app.post("/api/v1/search")
# async def search(q: str = Query(...), limit: int = 10):
#     with ix.searcher() as s:
#         parser = QueryParser("content", ix.schema)
#         results = s.search(parser.parse(q), limit=limit)
#         return {"hits": [dict(h) for h in results], "total": len(results)}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

4. Démarrer le serveur

uvicorn main:app --reload --port 8000

5. Tester l'API

# Vérification de santé
curl http://localhost:8000/api/v1/health

# Recherche
curl -X POST http://localhost:8000/api/v1/search \
  -H "Content-Type: application/json" \
  -d '{"q": "python recherche"}'

# Get document by ID
curl http://localhost:8000/api/v1/documents/doc1

# Autocomplétion
curl "http://localhost:8000/api/v1/autocomplete?q=py"

6. Indexation en lot

# Ajouter à main.py pour l'indexation dynamique
from fastapi import FastAPI
from whoosh.writing import BufferedWriter

@app.post("/api/v1/index")
async def index_docs(docs: list[dict]):
    with BufferedWriter(ix, period=30, limit=50) as w:
        for doc in docs:
            w.add_document(**doc)
    return {"indexed": len(docs)}

Points clés

• create_app() de whoosh_fastapi fournit les endpoints /health, /search et /autocomplete.
• Tous les appels bloquants s'exécutent hors boucle d'événements via run_sync.
• Utilisez BufferedWriter pour l'indexation en masse.
• WhooshFastAPI classe offre une enregistrement par endpoint pour les intégrations personnalisées.