Coverage for src / lexigram / admin / cli / commands / search.py: 0%

1from __future__ import annotations 

2 

3import asyncio 

4import importlib 

5from pathlib import Path 

6from typing import Annotated, Any 

7 

8import typer 

9 

10 

11def create_app(): 

12 """Create the search CLI command group (Typer app). 

13 

14 Called by ``lexigram-cli``'s command loader. 

15 """ 

16 

17 app = typer.Typer(name="search", help="Search index management") 

18 

19 @app.command() 

20 def reindex( 

21 config_path: Annotated[ 

22 Path, 

23 typer.Argument( 

24 help="Path to application.yaml (default: ./application.yaml)", 

25 ), 

26 ] = Path("application.yaml"), 

27 app_module: Annotated[ 

28 str | None, 

29 typer.Option( 

30 "--app-module", 

31 "-m", 

32 help="Module path to root module, e.g. 'mypackage.root:RootModule'", 

33 ), 

34 ] = None, 

35 ) -> None: 

36 """Reindex all searchable resources. 

37 

38 Boots the application from *config_path*, discovers all admin resources 

39 with a ``searchable`` spec, queries every record from each resource's 

40 data source, and indexes them through the configured search engine. 

41 """ 

42 asyncio.run(_run_reindex(config_path=config_path, app_module=app_module)) 

43 

44 return app 

45 

46 

47async def _run_reindex(config_path: Path, app_module: str | None) -> None: 

48 from typer import Exit as TyperExit 

49 from typer import echo 

50 

51 from lexigram.app import Application 

52 from lexigram.config.loader import ConfigLoader # type: ignore[import-untyped] 

53 from lexigram.config.main import LexigramConfig 

54 

55 if app_module is None: 

56 echo("Error: --app-module is required", err=True) 

57 raise TyperExit(1) 

58 

59 config = ConfigLoader().load_sync(LexigramConfig, config_path) 

60 

61 module_path, _, class_name = app_module.rpartition(":") 

62 try: 

63 mod = importlib.import_module(module_path) 

64 except ImportError as exc: 

65 echo(f"Error: cannot import module '{module_path}': {exc}", err=True) 

66 raise TyperExit(1) from None 

67 root_cls = getattr(mod, class_name, None) 

68 if root_cls is None: 

69 echo(f"Error: class '{class_name}' not found in '{module_path}'", err=True) 

70 raise TyperExit(1) 

71 

72 echo("Booting application...") 

73 dm = root_cls.configure(config=config) 

74 

75 try: 

76 async with Application.boot(modules=[dm], config=config) as app: 

77 echo("Collecting searchable resources...") 

78 searchable = await _collect_searchable(app.container) 

79 if not searchable: 

80 echo("No searchable resources found.") 

81 return 

82 

83 _log_banner(searchable) 

84 

85 from lexigram.search.engine import SearchEngine 

86 

87 search_engine = await app.container.resolve( 

88 SearchEngine, 

89 bypass_visibility=True, 

90 ) 

91 

92 total = 0 

93 for resource_cls, searchable_spec in searchable: 

94 dsc = getattr(resource_cls, "_data_source_class", None) 

95 if dsc is None: 

96 _log_warn(resource_cls, "no _data_source_class") 

97 continue 

98 try: 

99 ds = await app.container.resolve(dsc, bypass_visibility=True) 

100 except Exception as exc: 

101 _log_warn(resource_cls, f"resolve failed: {exc}") 

102 continue 

103 

104 repo = getattr(ds, "_repo", None) 

105 if repo is None: 

106 _log_warn(resource_cls, "no _repo on data source") 

107 continue 

108 

109 echo(f" Fetching {searchable_spec.index_name}...") 

110 try: 

111 records = await repo.find(limit=None) 

112 except Exception as exc: 

113 _log_warn(resource_cls, f"find failed: {exc}") 

114 continue 

115 

116 if not records: 

117 echo(" -> 0 records") 

118 continue 

119 

120 docs = _build_documents(records, searchable_spec) 

121 try: 

122 await search_engine.index(searchable_spec.index_name, docs) 

123 except Exception as exc: 

124 _log_warn(resource_cls, f"index failed: {exc}") 

125 continue 

126 

127 echo(f" -> indexed {len(docs)} document(s)") 

128 total += len(docs) 

129 

130 echo(f"\nReindex complete. {total} document(s) indexed.") 

131 except Exception as exc: 

132 echo(f"Error during reindex: {exc}", err=True) 

133 raise TyperExit(1) from None 

134 

135 

136async def _collect_searchable( 

137 container: object, 

138) -> list[tuple[type, Any]]: 

139 """Return (resource_class, SearchableSpec) pairs from the admin registry.""" 

140 from lexigram.admin.contributors.registry import ContributorRegistry 

141 from lexigram.admin.contributors.resource_collector import ResourceCollector 

142 from lexigram.admin.dashboard.naming_policy import NamingPolicy 

143 

144 try: 

145 registry = await container.resolve( # type: ignore[attr-defined] 

146 ContributorRegistry, 

147 bypass_visibility=True, 

148 ) 

149 except Exception: 

150 return [] 

151 

152 naming = NamingPolicy(mode="warn") 

153 collector = ResourceCollector(naming_policy=naming) 

154 all_contributors = list(registry.get_all()) 

155 resource_classes = collector.collect(all_contributors) 

156 

157 result: list[tuple[type, object]] = [] 

158 for rc in resource_classes: 

159 spec = getattr(rc, "searchable", None) 

160 if spec is not None and getattr(spec, "index_name", None): 

161 result.append((rc, spec)) 

162 return result 

163 

164 

165def _build_documents( 

166 records: list[object], 

167 searchable: object, 

168) -> list[dict]: 

169 """Build search document dicts from *records*. 

170 

171 The spec is expected to have a ``fields`` iterable. Each record is 

172 converted using ``getattr`` (object) or ``get`` (dict) style access. 

173 """ 

174 fields: tuple[str, ...] = getattr(searchable, "fields", ()) 

175 docs: list[dict] = [] 

176 for record in records: 

177 if isinstance(record, dict): 

178 doc_id = record.get("id") 

179 doc = {f: record.get(f) for f in fields} 

180 else: 

181 doc_id = getattr(record, "id", None) 

182 doc = {f: getattr(record, f, None) for f in fields} 

183 if doc_id is None: 

184 continue 

185 doc["id"] = str(doc_id) 

186 docs.append(doc) 

187 return docs 

188 

189 

190def _log_banner(searchable: list[tuple[type, object]]) -> None: 

191 from typer import echo 

192 

193 echo(f" Found {len(searchable)} searchable resource(s):") 

194 for rc, spec in searchable: 

195 echo(f" - {rc.__name__} -> {getattr(spec, 'index_name', '?')}") 

196 

197 

198def _log_warn(resource_cls: type, reason: str) -> None: 

199 from typer import echo 

200 

201 echo(f" Skipping {resource_cls.__name__}: {reason}")