""" Library Desk - Main FastAPI Application Following best practices: - Async routes for I/O operations - Dependency injection for configuration - Proper error handling - OpenAPI documentation """ from fastapi import FastAPI, HTTPException, Depends from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from typing import Dict, Any import logging from pathlib import Path from src.config import Settings, get_settings, __version__ from src.core.dependencies import verify_api_key # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) # Initialize FastAPI app app = FastAPI( title="Library Desk API", description="Coordination service for The Library system - HybridRAG queries, document ingestion, entity extraction, and mind map generation", version=__version__, docs_url="/docs", redoc_url="/redoc", ) # CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], # Configure appropriately for production allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Register routers from src.routers import ( wiki, tools, graph, vector, hybrid_rag, consolidation, ingestion, entity_linking, webhooks, rag_search, content ) app.include_router(wiki.router) app.include_router(tools.router) app.include_router(graph.router) app.include_router(vector.router) app.include_router(hybrid_rag.router) app.include_router(consolidation.router) app.include_router(ingestion.router) app.include_router(entity_linking.router) app.include_router(webhooks.router) app.include_router(rag_search.router) app.include_router(content.router) # Mount static files directory for Wiki.js integration scripts static_dir = Path(__file__).parent.parent / "static" if static_dir.exists(): app.mount("/static", StaticFiles(directory=str(static_dir)), name="static") logger.info(f"Mounted static files from {static_dir}") # Response Models class HealthResponse(BaseModel): """Health check response model.""" status: str app_name: str version: str services: Dict[str, Any] class StatsResponse(BaseModel): """Statistics response model.""" wiki_pages: int neo4j_nodes: int qdrant_vectors: int # Routes @app.get("/", tags=["Root"]) async def root() -> Dict[str, str]: """Root endpoint.""" return { "message": "Library Desk API", "docs": "/docs", "health": "/health" } @app.get("/health", response_model=HealthResponse, tags=["System"]) async def health(settings: Settings = Depends(get_settings)) -> HealthResponse: """ Health check endpoint. Returns status of all connected services. """ from src.core.dependencies import check_service_health # Check service connectivity service_health = await check_service_health() # Overall status is healthy if at least Neo4j and Qdrant are up all_healthy = service_health.get("neo4j", False) and service_health.get("qdrant", False) overall_status = "healthy" if all_healthy else "degraded" return HealthResponse( status=overall_status, app_name=settings.app_name, version=settings.app_version, services={ "neo4j": { "url": settings.neo4j_uri, "healthy": service_health.get("neo4j", False) }, "qdrant": { "url": settings.qdrant_url, "healthy": service_health.get("qdrant", False) }, "wikijs": { "url": settings.wikijs_url, "healthy": service_health.get("wikijs", False) }, "searxng": { "url": settings.searxng_url, "healthy": service_health.get("searxng", False) }, "ollama": { "url": settings.ollama_url, "model": settings.ollama_model, "healthy": service_health.get("ollama", False) } } ) @app.get("/stats", response_model=StatsResponse, tags=["System"]) async def stats( api_key: str = Depends(verify_api_key) ) -> StatsResponse: """ Get system statistics. Protected endpoint - requires API key. TODO: Implement actual stats gathering from: - Neo4j (node count) - Qdrant (vector count) - Wiki.js (page count) """ return StatsResponse( wiki_pages=0, neo4j_nodes=0, qdrant_vectors=0 ) # Ingestion endpoints (for Scheduler integration) @app.post("/ingest/document", tags=["Ingestion"]) async def ingest_document( document: Dict[str, Any], api_key: str = Depends(verify_api_key) ) -> Dict[str, Any]: """ Ingest a single document for indexing. Used by The Scheduler to add mirrored documentation to the knowledge base. Expected fields: - source: str (e.g., "github", "gitea") - repository: str (e.g., "anthropic-cookbook") - path: str (file path) - content: str (document content) - metadata: dict (commit, author, tags, etc.) TODO: Implement document ingestion pipeline: 1. Chunk content 2. Generate embeddings (Ollama) 3. Extract entities (NLP) 4. Index in Qdrant 5. Create graph nodes/relationships in Neo4j """ return { "message": "Document ingestion not yet implemented", "document_id": f"doc_{document.get('path', 'unknown')}", "status": "stub" } @app.post("/ingest/batch", tags=["Ingestion"]) async def batch_ingest( batch: Dict[str, Any], api_key: str = Depends(verify_api_key) ) -> Dict[str, Any]: """ Ingest multiple documents in a batch. More efficient than individual ingestion for large syncs. TODO: Implement batch processing with task queue """ document_count = len(batch.get("documents", [])) return { "message": "Batch ingestion not yet implemented", "batch_id": "batch_stub", "total_documents": document_count, "status": "stub" } @app.post("/ingest/check-updates", tags=["Ingestion"]) async def check_updates( documents: Dict[str, Any], api_key: str = Depends(verify_api_key) ) -> Dict[str, Any]: """ Check which documents need updating based on content hashes. Used by Scheduler to determine what changed since last sync. TODO: Implement update detection: 1. Query existing documents by path 2. Compare content hashes 3. Return list of updates needed """ return { "message": "Update checking not yet implemented", "updates_needed": [], "up_to_date": [], "new_documents": [] } @app.get("/ingest/status/{document_id}", tags=["Ingestion"]) async def get_ingestion_status( document_id: str, api_key: str = Depends(verify_api_key) ) -> Dict[str, Any]: """ Get processing status for a document. TODO: Implement status tracking """ return { "message": "Status tracking not yet implemented", "document_id": document_id, "status": "unknown" } @app.get("/ingest/repo-status/{repository}", tags=["Ingestion"]) async def get_repo_status( repository: str, api_key: str = Depends(verify_api_key) ) -> Dict[str, Any]: """ Get indexing status for an entire repository. TODO: Implement repository-level statistics """ return { "message": "Repository status not yet implemented", "repository": repository, "total_documents": 0, "indexed_documents": 0 } # Query endpoints (stubs for future implementation) # NOTE: /query/hybrid is now implemented in routers/hybrid_rag.py @app.post("/query/semantic", tags=["Query"]) async def semantic_query( query: Dict[str, Any], api_key: str = Depends(verify_api_key) ) -> Dict[str, Any]: """ Semantic search via Qdrant. Pure vector similarity search. TODO: Implement semantic search """ return { "message": "Semantic search not yet implemented", "query": query } @app.post("/query/graph", tags=["Query"]) async def graph_query( query: Dict[str, Any], api_key: str = Depends(verify_api_key) ) -> Dict[str, Any]: """ Graph traversal via Neo4j. Execute Cypher queries. TODO: Implement graph queries """ return { "message": "Graph query not yet implemented", "query": query } # Deduplication endpoints @app.post("/deduplicate/check", tags=["Deduplication"]) async def check_duplicates( request: Dict[str, Any], api_key: str = Depends(verify_api_key) ) -> Dict[str, Any]: """ Check for duplicate or highly similar documents. Uses vector similarity and graph analysis. Expected fields: - document_id: str - similarity_threshold: float (default 0.85) TODO: Implement deduplication: 1. Get document embedding from Qdrant 2. Find similar vectors above threshold 3. Check graph relationships 4. Return candidates with similarity scores """ document_id = request.get("document_id") threshold = request.get("similarity_threshold", 0.85) return { "message": "Deduplication not yet implemented", "document_id": document_id, "threshold": threshold, "duplicates": [], "suggestions": None } # Application lifecycle @app.on_event("startup") async def startup_event(): """Initialize connections and resources on startup.""" from src.core.dependencies import startup_clients from src.services.wiki_change_listener import WikiChangeListener settings = get_settings() logger.info(f"Starting {settings.app_name} v{settings.app_version}") logger.info(f"Neo4j: {settings.neo4j_uri}") logger.info(f"Qdrant: {settings.qdrant_url}") logger.info(f"Wiki.js: {settings.wikijs_url}") logger.info(f"SearXNG: {settings.searxng_url}") logger.info(f"Ollama: {settings.ollama_url}") # Initialize all service clients await startup_clients() # Start Wiki.js change listener (PostgreSQL NOTIFY/LISTEN) # This enables automatic processing of user-edited pages try: wiki_listener = WikiChangeListener() await wiki_listener.start() # Store reference for shutdown app.state.wiki_listener = wiki_listener logger.info("Wiki.js change listener started successfully") except Exception as e: logger.error(f"Failed to start Wiki.js change listener: {e}", exc_info=True) logger.warning("Continuing without change listener - manual page updates will not be auto-processed") @app.on_event("shutdown") async def shutdown_event(): """Clean up resources on shutdown.""" from src.core.dependencies import shutdown_clients logger.info("Shutting down Library Desk API") # Stop Wiki.js change listener if running if hasattr(app.state, "wiki_listener"): try: await app.state.wiki_listener.stop() logger.info("Wiki.js change listener stopped") except Exception as e: logger.error(f"Error stopping Wiki.js change listener: {e}") # Close all service clients await shutdown_clients()