""" Dependency injection for Library Desk. Provides FastAPI dependencies for service clients with: - Singleton pattern via @lru_cache - Lazy initialization - Proper lifecycle management - Type aliases for clean endpoint signatures """ import asyncio from functools import lru_cache from typing import TYPE_CHECKING, Annotated from fastapi import Depends, HTTPException, Query import logging import redis.asyncio as aioredis import secrets from fastapi import Request, Security from fastapi.security import HTTPBearer from src.config import Settings, get_settings from src.clients.neo4j_client import Neo4jClient from src.clients.qdrant_client import QdrantClientWrapper from src.clients.wikijs_client import WikiJSClient from src.clients.searxng_client import SearXNGClient from src.clients.ollama_client import OllamaClient from src.clients.content_extractor import ContentExtractor from src.clients.paperless_client import PaperlessClient from src.clients.settings_client import SettingsClient from src.clients.scheduler_client import SchedulerClient from src.apis import ( OpenMeteoProvider, AggregatedNewsProvider, AlphaVantageProvider, ) logger = logging.getLogger(__name__) # Settings dependency SettingsDep = Annotated[Settings, Depends(get_settings)] # Tenant user dependency def require_user( user: str = Query( ..., description=( "User identifier (tenant). Required — every operation is scoped to " "this tenant's namespace (Qdrant collection, Neo4j labels, wiki path, " "Redis keys). Requests without an explicit non-empty user are " "rejected with 422. There is no default tenant." ), ) ) -> str: """ FastAPI dependency: required tenant user query parameter. Rejects missing (FastAPI returns 422 automatically), empty, and whitespace-only user values. Use via the RequiredUserQuery alias. """ from src.core.multi_tenancy import validate_required_user try: return validate_required_user(user) except ValueError as e: raise HTTPException(status_code=422, detail=str(e)) RequiredUserQuery = Annotated[str, Depends(require_user)] # Client factory functions with @lru_cache for singletons @lru_cache def get_neo4j_client() -> Neo4jClient: """ Get Neo4j client singleton. Returns: Initialized Neo4j client (not yet connected) Note: Call client.connect() during app startup """ settings = get_settings() client = Neo4jClient( uri=settings.neo4j_uri, user=settings.neo4j_user, password=settings.neo4j_password ) logger.debug("Created Neo4j client instance") return client @lru_cache def get_qdrant_client() -> QdrantClientWrapper: """ Get Qdrant client singleton. Returns: Initialized Qdrant client Note: Collections are created lazily per-user """ settings = get_settings() client = QdrantClientWrapper( url=settings.qdrant_url, embedding_dim=768, # nomic-embed-text default timeout=settings.qdrant_timeout ) logger.debug("Created Qdrant client instance") return client @lru_cache def get_wikijs_client() -> WikiJSClient: """ Get Wiki.js client singleton. Returns: Initialized Wiki.js GraphQL client with API token auth """ settings = get_settings() client = WikiJSClient( base_url=settings.wikijs_url, api_token=settings.wiki_graphql_api ) logger.debug("Created Wiki.js client instance") return client @lru_cache def get_searxng_client() -> SearXNGClient: """ Get SearXNG client singleton. Returns: Initialized SearXNG search client """ settings = get_settings() client = SearXNGClient(base_url=settings.searxng_url) logger.debug("Created SearXNG client instance") return client @lru_cache def get_ollama_client() -> OllamaClient: """ Get Ollama client singleton. Returns: Initialized Ollama embeddings client """ settings = get_settings() client = OllamaClient( base_url=settings.ollama_url, model=settings.ollama_embedding_model ) logger.debug("Created Ollama client instance") return client @lru_cache def get_redis_client() -> aioredis.Redis: """ Get Redis client singleton for caching. Returns: Async Redis client connected to the configured database Note: Uses Redis DB 4 (configured for library-desk) """ settings = get_settings() client = aioredis.from_url( settings.redis_url, encoding="utf-8", decode_responses=True ) logger.debug(f"Created Redis client: {settings.redis_url}") return client @lru_cache def get_job_manager() -> "JobManager": """ Get Redis-backed JobManager singleton. Returns: JobManager for background job tracking (connects lazily) """ from src.jobs.job_manager import JobManager settings = get_settings() manager = JobManager(redis_url=settings.redis_url) logger.debug(f"Created JobManager: {settings.redis_url}") return manager @lru_cache def get_content_extractor() -> ContentExtractor: """ Get ContentExtractor singleton. Returns: Initialized content extraction client using Trafilatura """ settings = get_settings() extractor = ContentExtractor( timeout=settings.content_extraction_timeout, max_length=settings.content_max_length ) logger.debug("Created ContentExtractor instance") return extractor @lru_cache def get_paperless_client() -> PaperlessClient: """ Get Paperless-ngx client singleton. Returns: Initialized Paperless-ngx REST API client Note: Returns None-like client if paperless_token is not configured """ settings = get_settings() if not settings.paperless_token: logger.warning("Paperless token not configured - document storage disabled") client = PaperlessClient( base_url=settings.paperless_url, token=settings.paperless_token, timeout=settings.paperless_timeout ) logger.debug(f"Created Paperless client: {settings.paperless_url}") return client @lru_cache def get_settings_client() -> SettingsClient: """ Get central settings database client singleton. Returns: Initialized SettingsClient for Tatlock system_settings database Note: Returns client with empty DSN if password not configured """ settings = get_settings() if not settings.system_settings_password: logger.warning("System settings password not configured - settings database disabled") client = SettingsClient(dsn=settings.system_settings_dsn) logger.debug(f"Created Settings client: {settings.system_settings_host}") return client @lru_cache def get_scheduler_client() -> SchedulerClient: """ Get scheduler service client singleton. Returns: Initialized SchedulerClient for task management Note: Used for registering prefetch tasks discovered during HybridRAG searches """ settings = get_settings() client = SchedulerClient( base_url=settings.scheduler_url, api_key=settings.scheduler_api_key, ) logger.debug(f"Created Scheduler client: {settings.scheduler_url}") return client # ============================================================================= # External API Providers # ============================================================================= @lru_cache def get_weather_provider() -> OpenMeteoProvider: """ Get Open-Meteo weather provider singleton. Returns: Initialized OpenMeteoProvider with default timezone Note: Timezone can be overridden per-request for user preferences """ provider = OpenMeteoProvider(timezone="Europe/Amsterdam") logger.debug("Created OpenMeteo weather provider") return provider # News provider requires sources from settings database _news_provider: AggregatedNewsProvider | None = None async def get_news_provider() -> AggregatedNewsProvider: """ Get aggregated news provider. Returns: Initialized AggregatedNewsProvider with user-configured sources and per-source category filters. Note: Configuration is fetched from system_settings database: - news.sources: list of enabled sources (default: ["nos", "bbc"]) - api.{source}.categories: list of enabled categories per source """ global _news_provider if _news_provider is not None: return _news_provider settings_client = get_settings_client() # Get enabled sources sources = await settings_client.get("news.sources") if not sources or not isinstance(sources, list): sources = ["nos", "bbc"] logger.info(f"Using default news sources: {sources}") else: logger.info(f"Using configured news sources: {sources}") # Filter out disabled sources and get category filters enabled_sources: list[str] = [] category_filters: dict[str, list[str]] = {} for source in sources: config = await settings_client.get_api_config(source) if config: # Check if source is disabled if config.get("enabled") is False: logger.info(f"News source '{source}' is disabled - skipping") continue # Get category filter if specified categories = config.get("categories", []) if categories: category_filters[source] = categories logger.debug(f"Source '{source}' categories: {categories}") enabled_sources.append(source) if not enabled_sources: enabled_sources = ["nos", "bbc"] logger.warning("No enabled news sources - using defaults") _news_provider = AggregatedNewsProvider( sources=enabled_sources, category_filters=category_filters ) return _news_provider # AlphaVantage requires API key from settings database _alphavantage_provider: AlphaVantageProvider | None = None async def get_alphavantage_provider() -> AlphaVantageProvider | None: """ Get Alpha Vantage financial provider. Returns: Initialized AlphaVantageProvider or None if API key not configured Note: API key is fetched from system_settings database """ global _alphavantage_provider if _alphavantage_provider is not None: return _alphavantage_provider settings_client = get_settings_client() api_key = await settings_client.get_api_key("alphavantage") if not api_key: logger.warning("Alpha Vantage API key not configured - financial provider disabled") return None _alphavantage_provider = AlphaVantageProvider(api_key=api_key) logger.debug("Created Alpha Vantage financial provider") return _alphavantage_provider # Type aliases for FastAPI endpoint dependencies # Usage: def my_endpoint(neo4j: Neo4jDep): Neo4jDep = Annotated[Neo4jClient, Depends(get_neo4j_client)] QdrantDep = Annotated[QdrantClientWrapper, Depends(get_qdrant_client)] WikiJSDep = Annotated[WikiJSClient, Depends(get_wikijs_client)] SearXNGDep = Annotated[SearXNGClient, Depends(get_searxng_client)] OllamaDep = Annotated[OllamaClient, Depends(get_ollama_client)] RedisDep = Annotated[aioredis.Redis, Depends(get_redis_client)] ContentExtractorDep = Annotated[ContentExtractor, Depends(get_content_extractor)] PaperlessDep = Annotated[PaperlessClient, Depends(get_paperless_client)] SettingsClientDep = Annotated[SettingsClient, Depends(get_settings_client)] SchedulerDep = Annotated[SchedulerClient, Depends(get_scheduler_client)] from src.jobs.job_manager import JobManager # noqa: E402 JobManagerDep = Annotated[JobManager, Depends(get_job_manager)] # External API provider dependencies WeatherProviderDep = Annotated[OpenMeteoProvider, Depends(get_weather_provider)] NewsProviderDep = Annotated[AggregatedNewsProvider, Depends(get_news_provider)] AlphaVantageProviderDep = Annotated[AlphaVantageProvider | None, Depends(get_alphavantage_provider)] # Lifecycle management functions async def startup_clients(): """ Initialize all service clients at application startup. Should be called in FastAPI lifespan or startup event. Performs: - Neo4j connection pool initialization - Neo4j connectivity verification - Ollama model availability check """ logger.info("Starting up service clients...") # Initialize Neo4j connection pool neo4j = get_neo4j_client() try: await neo4j.connect() logger.info("✓ Neo4j connected") except Exception as e: logger.error(f"✗ Neo4j connection failed: {e}") # Don't fail startup - allow degraded operation pass # Check Ollama availability ollama = get_ollama_client() try: is_healthy = await ollama.health_check() if is_healthy: logger.info(f"✓ Ollama ready (model: {ollama.model})") else: logger.warning(f"✗ Ollama model '{ollama.model}' not available") except Exception as e: logger.error(f"✗ Ollama health check failed: {e}") pass # Check Paperless availability settings = get_settings() if settings.paperless_token: try: paperless = get_paperless_client() is_healthy = await paperless.health_check() if is_healthy: logger.info(f"✓ Paperless-ngx ready: {settings.paperless_url}") else: logger.warning("✗ Paperless-ngx not responding") except Exception as e: logger.error(f"✗ Paperless health check failed: {e}") else: logger.info("○ Paperless-ngx not configured (document storage disabled)") # Check System Settings database availability if settings.system_settings_password: try: settings_client = get_settings_client() is_healthy = await settings_client.health_check() if is_healthy: logger.info(f"✓ System settings DB ready: {settings.system_settings_host}") else: logger.warning("✗ System settings DB not responding") except Exception as e: logger.error(f"✗ System settings health check failed: {e}") else: logger.info("○ System settings not configured") # Check Scheduler availability try: scheduler = get_scheduler_client() is_healthy = await scheduler.health_check() if is_healthy: logger.info(f"✓ Scheduler ready: {settings.scheduler_url}") else: logger.warning("✗ Scheduler not responding") except Exception as e: logger.error(f"✗ Scheduler health check failed: {e}") # Qdrant, Wiki.js, SearXNG are lazy-initialized logger.info("Service clients startup complete") async def shutdown_clients(): """ Cleanup all service clients at application shutdown. Should be called in FastAPI lifespan or shutdown event. Performs: - Close Neo4j connection pool - Close HTTP clients """ logger.info("Shutting down service clients...") # Close Neo4j driver neo4j = get_neo4j_client() try: await neo4j.close() logger.info("✓ Neo4j closed") except Exception as e: logger.error(f"Error closing Neo4j: {e}") # Close HTTP clients clients_to_close = [ ("Wiki.js", get_wikijs_client()), ("SearXNG", get_searxng_client()), ("Ollama", get_ollama_client()), ("Paperless", get_paperless_client()), ("OpenMeteo", get_weather_provider()), ] for name, client in clients_to_close: try: await client.close() logger.info(f"✓ {name} client closed") except Exception as e: logger.error(f"Error closing {name} client: {e}") # Close async-initialized providers global _news_provider, _alphavantage_provider if _news_provider is not None: try: await _news_provider.close() _news_provider = None logger.info("✓ News provider closed") except Exception as e: logger.error(f"Error closing News provider: {e}") if _alphavantage_provider is not None: try: await _alphavantage_provider.close() _alphavantage_provider = None logger.info("✓ AlphaVantage client closed") except Exception as e: logger.error(f"Error closing AlphaVantage client: {e}") # Close settings database connection settings = get_settings() if settings.system_settings_password: try: settings_client = get_settings_client() await settings_client.close() logger.info("✓ System settings client closed") except Exception as e: logger.error(f"Error closing settings client: {e}") # Close scheduler client try: scheduler = get_scheduler_client() await scheduler.close() logger.info("✓ Scheduler client closed") except Exception as e: logger.error(f"Error closing scheduler client: {e}") # Close job manager Redis connection try: job_manager = get_job_manager() await job_manager.close() logger.info("✓ JobManager closed") except Exception as e: logger.error(f"Error closing JobManager: {e}") logger.info("Service clients shutdown complete") # Per-probe timeout for the concurrent health checks below, in seconds. # # Bounded well under the container healthcheck's 10s timeout (see Dockerfile). # Because every probe runs concurrently under one asyncio.gather, the wall time # is one bound rather than the sum, so adding probes does not erode the margin — # all eight can hang and the call still returns in ~2s. # # A probe with no bound of its own falls back to its client's default: 30s for # the neo4j and qdrant drivers, 30s for the scheduler client, and none at all # for system settings. Each is past the 10s budget on its own, which is what let # a hung dependency — not a failing one — flip the container unhealthy. HEALTH_PROBE_TIMEOUT = 2.0 async def _probe_neo4j() -> bool: """Neo4j connectivity probe. Returns True if healthy.""" try: neo4j = get_neo4j_client() # Simple query to check connectivity await neo4j.execute_query("RETURN 1 as test", {}) return True except Exception as e: logger.error(f"Neo4j health check failed: {e}") return False async def _probe_qdrant() -> bool: """Qdrant connectivity probe. Returns True if healthy.""" try: qdrant = get_qdrant_client() # Check if we can list collections await qdrant.client.get_collections() return True except Exception as e: logger.error(f"Qdrant health check failed: {e}") return False async def _probe_wikijs() -> bool: """Wiki.js connectivity probe. Returns True if healthy.""" try: wikijs = get_wikijs_client() # Try a simple query (list pages with limit 1) await wikijs.list_pages(limit=1) return True except Exception as e: logger.error(f"Wiki.js health check failed: {e}") return False async def _probe_searxng() -> bool: """SearXNG connectivity probe. Returns True if healthy.""" try: searxng = get_searxng_client() # Just check if service is up (no actual search) return await searxng.health_check() except Exception as e: logger.error(f"SearXNG health check failed: {e}") return False async def _probe_ollama() -> bool: """Ollama connectivity probe. Returns True if healthy.""" try: ollama = get_ollama_client() return await ollama.health_check() except Exception as e: logger.error(f"Ollama health check failed: {e}") return False async def _probe_paperless() -> bool: """Paperless-ngx connectivity probe. Returns True if healthy.""" try: paperless = get_paperless_client() return await paperless.health_check() except Exception as e: logger.error(f"Paperless health check failed: {e}") return False async def _probe_system_settings() -> bool: """System settings database connectivity probe. Returns True if healthy.""" try: settings_client = get_settings_client() return await settings_client.health_check() except Exception as e: logger.error(f"System settings health check failed: {e}") return False async def _probe_scheduler() -> bool: """Scheduler connectivity probe. Returns True if healthy.""" try: scheduler = get_scheduler_client() return await scheduler.health_check() except Exception as e: logger.error(f"Scheduler health check failed: {e}") return False async def _bounded_probe(coro) -> bool: """ Run a probe coroutine bounded by HEALTH_PROBE_TIMEOUT. A probe that times out is reported unhealthy, the same as one that raises. Wrapping happens here rather than in each _probe_* function so the bound applies uniformly regardless of whether the underlying client has its own (looser, or absent) timeout. """ try: return await asyncio.wait_for(coro, timeout=HEALTH_PROBE_TIMEOUT) except asyncio.TimeoutError: logger.error(f"Health probe timed out after {HEALTH_PROBE_TIMEOUT}s") return False async def check_service_health() -> dict: """ Check health of all service clients. All eight probes (neo4j, qdrant, wikijs, searxng, ollama, paperless, system_settings, scheduler) run concurrently in one gather, each bounded at HEALTH_PROBE_TIMEOUT, so a single hung dependency cannot block the others or push the endpoint past the container healthcheck's timeout. return_exceptions=True means one probe raising cannot cancel its siblings. paperless and system_settings are three-state: None means "not configured" (no probe is run at all — a probe never joins the gather for a service that has no credentials to check), False means configured but unreachable/unhealthy (including a timeout), True means healthy. Collapsing "not configured" into "unhealthy" would be a different claim than the one this function is making, so that decision is made before the gather rather than by feeding an unconfigured probe through the same bool-returning bound as the rest. Only neo4j, qdrant, wikijs, searxng and ollama are read by /health (src/main.py) — paperless, system_settings and scheduler are computed here but not currently surfaced by any caller (checked: the only two callers are src/main.py and tests/test_integration.py, and the test asserts only the five). Bounded rather than removed, since bounding cannot break a hypothetical consumer and deleting could. Returns: Dictionary with health status of each service: { "neo4j": bool, "qdrant": bool, "wikijs": bool, "searxng": bool, "ollama": bool, "paperless": bool | None, "system_settings": bool | None, "scheduler": bool } Usage: >>> health = await check_service_health() >>> health["neo4j"] True """ health = {} settings = get_settings() # Build the probe list dynamically: paperless and system_settings only # join it when configured, so an unconfigured service is never bounded, # timed out, or reported False — it is set to None directly, below. probe_names = ["neo4j", "qdrant", "wikijs", "searxng", "ollama"] probes = [ _bounded_probe(_probe_neo4j()), _bounded_probe(_probe_qdrant()), _bounded_probe(_probe_wikijs()), _bounded_probe(_probe_searxng()), _bounded_probe(_probe_ollama()), ] if settings.paperless_token: probe_names.append("paperless") probes.append(_bounded_probe(_probe_paperless())) else: health["paperless"] = None # Not configured if settings.system_settings_password: probe_names.append("system_settings") probes.append(_bounded_probe(_probe_system_settings())) else: health["system_settings"] = None # Not configured # Scheduler has no config gate - it always runs. probe_names.append("scheduler") probes.append(_bounded_probe(_probe_scheduler())) results = await asyncio.gather(*probes, return_exceptions=True) # _bounded_probe already catches everything from its own probe, but # return_exceptions=True also guards against a bug in _bounded_probe # itself surfacing as an unhandled exception here. for name, result in zip(probe_names, results): if isinstance(result, BaseException): logger.error(f"{name} health check raised unexpectedly: {result}") health[name] = False else: health[name] = result return health # Service factory functions @lru_cache def get_vector_service() -> "VectorService": """Get VectorService singleton.""" from src.services.vector_service import VectorService return VectorService( qdrant_client=get_qdrant_client(), wikijs_client=get_wikijs_client(), ollama_client=get_ollama_client() ) @lru_cache def get_graph_service() -> "GraphService": """Get GraphService singleton.""" from src.services.graph_service import GraphService return GraphService( neo4j_client=get_neo4j_client(), wikijs_client=get_wikijs_client() ) @lru_cache def get_wiki_service() -> "WikiService": """Get WikiService singleton.""" from src.services.wiki_service import WikiService return WikiService(wiki_client=get_wikijs_client()) @lru_cache def get_consolidation_service() -> "ConsolidationService": """Get ConsolidationService singleton.""" from src.services.consolidation_service import ConsolidationService return ConsolidationService( neo4j=get_neo4j_client(), ollama=get_ollama_client(), wiki=get_wikijs_client(), settings=get_settings(), ingestion_service=get_ingestion_service(), volatile_service=get_volatile_cache_service(), settings_client=get_settings_client(), scheduler_client=get_scheduler_client(), ) @lru_cache def get_ingestion_service() -> "IngestionService": """Get IngestionService singleton.""" from src.services.ingestion_service import IngestionService return IngestionService( vector_service=get_vector_service(), graph_service=get_graph_service(), wiki_client=get_wikijs_client() ) @lru_cache def get_hybrid_rag_service() -> "HybridRAGService": """ Get HybridRAGService singleton. The single wiring point for HybridRAG — routers must depend on this instead of constructing their own instance (previous inline copies in the /query/hybrid and /wiki/smart-create routers diverged on volatile_service). """ from src.services.hybrid_rag_service import HybridRAGService return HybridRAGService( vector_service=get_vector_service(), graph_service=get_graph_service(), searxng_client=get_searxng_client(), ollama_client=get_ollama_client(), content_extractor=get_content_extractor(), settings=get_settings(), volatile_service=get_volatile_cache_service() ) @lru_cache def get_rag_search_service() -> "RAGSearchService": """Get RAGSearchService singleton.""" from src.services.rag_search_service import RAGSearchService return RAGSearchService( searxng_client=get_searxng_client(), content_extractor=get_content_extractor(), redis_client=get_redis_client(), settings=get_settings() ) @lru_cache def get_volatile_cache_service() -> "VolatileCacheService": """Get VolatileCacheService singleton.""" from src.services.volatile_service import VolatileCacheService return VolatileCacheService( qdrant_client=get_qdrant_client(), ollama_client=get_ollama_client(), settings=get_settings() ) security = HTTPBearer() async def verify_api_key( credentials: Annotated[HTTPBearer, Security(security)], settings: SettingsDep ) -> str: """ Verify API key from Bearer token. Args: credentials: HTTP Bearer credentials settings: Application settings Returns: API key if valid Raises: HTTPException: If API key is invalid """ if not secrets.compare_digest(credentials.credentials, settings.library_api_key): raise HTTPException( status_code=403, detail="Invalid API key" ) return credentials.credentials # Header set by NPM only on the authenticated /library-desk/ proxy location. # library-desk is bound to loopback (127.0.0.1:8089), so NPM is the only path # that can reach it and set this header — a client cannot forge it. NPM also # overwrites any client-supplied value via proxy_set_header. _PROXY_MARKER_HEADER = "x-library-desk-proxy" async def verify_browser_request( request: Request, settings: SettingsDep, ) -> str: """ Auth for browser-facing endpoints (the Wiki.js integration buttons). Accepts the request when it arrives through the authenticated NPM proxy location (Authentik session for external users, or the LAN bypass for internal ones) — identified by the trusted proxy marker header. No secret is carried in the browser. Machine callers may still authenticate with the Bearer API key. Returns the acting user's identity. """ if request.headers.get(_PROXY_MARKER_HEADER) == "1": # Authentik injects the identity for externally-authenticated users; # on the LAN bypass these are empty and the endpoint falls back to the # user supplied in the request body. return request.headers.get("x-authentik-email") or "lan" # Fallback: server-to-server Bearer API key. auth = request.headers.get("authorization", "") if auth.startswith("Bearer ") and secrets.compare_digest( auth[len("Bearer "):], settings.library_api_key ): return auth[len("Bearer "):] raise HTTPException(status_code=401, detail="Unauthenticated") # Service type aliases for FastAPI endpoint dependencies. # Deliberately imported here rather than at the top: these modules import back # into this one, so a module-level import would cycle. The factory functions # above must exist before they are pulled in. from src.services.vector_service import VectorService # noqa: E402 from src.services.graph_service import GraphService # noqa: E402 # Imported for annotations only. The real imports live inside the functions # that use them, to break an import cycle; a quoted annotation is never # evaluated at runtime, so the names were unresolvable to any checker. This # block costs nothing at import time and makes them resolvable again. if TYPE_CHECKING: from src.services.consolidation_service import ConsolidationService from src.services.hybrid_rag_service import HybridRAGService from src.services.ingestion_service import IngestionService from src.services.rag_search_service import RAGSearchService from src.services.volatile_service import VolatileCacheService from src.services.wiki_service import WikiService VectorServiceDep = Annotated[VectorService, Depends(get_vector_service)] GraphServiceDep = Annotated[GraphService, Depends(get_graph_service)]