Coverage for src / lilbee / services.py: 100%

1"""Typed service container — single point of access for all singletons. 

2 

3All runtime dependencies (provider, store, embedder, reranker, concepts, 

4clusterer, searcher) are created lazily on first call to ``get_services()`` 

5and cached for the process lifetime. Tests call ``reset_services()`` 

6between runs. 

7""" 

8 

9from __future__ import annotations 

10 

11import atexit 

12from dataclasses import dataclass 

13from typing import TYPE_CHECKING 

14 

15if TYPE_CHECKING: 

16 from lilbee.clustering import Clusterer 

17 from lilbee.concepts import ConceptGraph 

18 from lilbee.embedder import Embedder 

19 from lilbee.providers.base import LLMProvider 

20 from lilbee.query import Searcher 

21 from lilbee.registry import ModelRegistry 

22 from lilbee.reranker import Reranker 

23 from lilbee.store import Store 

24 

25 

26@dataclass(frozen=True) 

27class Services: 

28 """Holds all runtime service instances.""" 

29 

30 provider: LLMProvider 

31 store: Store 

32 embedder: Embedder 

33 reranker: Reranker 

34 concepts: ConceptGraph 

35 clusterer: Clusterer 

36 searcher: Searcher 

37 registry: ModelRegistry 

38 

39 

40_svc: Services | None = None 

41 

42 

43def get_services() -> Services: 

44 """Return the cached Services singleton, creating on first call. 

45 

46 Service modules are imported inside the function to keep CLI 

47 startup fast: ``services`` is on every CLI import path, and the 

48 concrete service modules transitively pull in heavy libraries 

49 (llama-cpp, lancedb, kreuzberg). Deferring the loads until first 

50 ``get_services()`` call makes ``lilbee --help`` and TUI splash 

51 render in milliseconds instead of seconds. 

52 """ 

53 global _svc 

54 if _svc is not None: 

55 return _svc 

56 

57 from lilbee.clustering import Clusterer 

58 from lilbee.concepts import ConceptGraph 

59 from lilbee.config import cfg 

60 from lilbee.embedder import Embedder 

61 from lilbee.providers.factory import create_provider 

62 from lilbee.query import Searcher 

63 from lilbee.registry import ModelRegistry 

64 from lilbee.reranker import Reranker 

65 from lilbee.store import Store 

66 

67 provider = create_provider(cfg) 

68 store = Store(cfg) 

69 embedder = Embedder(cfg, provider) 

70 reranker = Reranker(cfg) 

71 concepts = ConceptGraph(cfg, store) 

72 clusterer = Clusterer(cfg, store) 

73 registry = ModelRegistry(cfg.models_dir) 

74 searcher = Searcher(cfg, provider, store, embedder, reranker, concepts) 

75 _svc = Services( 

76 provider=provider, 

77 store=store, 

78 embedder=embedder, 

79 reranker=reranker, 

80 concepts=concepts, 

81 clusterer=clusterer, 

82 searcher=searcher, 

83 registry=registry, 

84 ) 

85 return _svc 

86 

87 

88def set_services(services: Services | None) -> None: 

89 """Replace the cached Services singleton (for testing).""" 

90 global _svc 

91 _svc = services 

92 

93 

94def reset_services() -> None: 

95 """Shut down and discard all cached instances.""" 

96 global _svc 

97 if _svc is not None: 

98 _svc.provider.shutdown() 

99 _svc.store.close() 

100 _svc = None 

101 

102 

103atexit.register(reset_services)