Coverage for src / ezqt_app / services / config / config_service.py: 48.79%

1# /////////////////////////////////////////////////////////////// 

2# SERVICES.CONFIG.CONFIG_SERVICE - Config service implementation 

3# Project: ezqt_app 

4# /////////////////////////////////////////////////////////////// 

5 

6"""Configuration service — full implementation (absorbs ConfigManager logic).""" 

7 

8from __future__ import annotations 

9 

10# /////////////////////////////////////////////////////////////// 

11# IMPORTS 

12# /////////////////////////////////////////////////////////////// 

13# Standard library imports 

14import shutil 

15import sys 

16from importlib.resources import files as _pkg_files 

17from pathlib import Path 

18from typing import Any 

19 

20# Third-party imports 

21import yaml 

22 

23# Local imports 

24from ...domain.ports.config_service import ConfigServiceProtocol 

25from ...utils.diagnostics import warn_user 

26from ...utils.printer import get_printer 

27from ...utils.runtime_paths import APP_PATH, get_bin_path 

28 

29 

30# /////////////////////////////////////////////////////////////// 

31# CLASSES 

32# /////////////////////////////////////////////////////////////// 

33class ConfigService(ConfigServiceProtocol): 

34 """Modular configuration service for EzQt_App.""" 

35 

36 def __init__(self): 

37 self._config_cache: dict[str, Any] = {} 

38 self._config_files: dict[str, Path] = {} 

39 self._project_root: Path | None = None 

40 

41 # ----------------------------------------------------------- 

42 # Port methods (ConfigServiceProtocol) 

43 # ----------------------------------------------------------- 

44 

45 def set_project_root(self, project_root: Path | str) -> None: 

46 """Set the active project root directory.""" 

47 self._project_root = ( 

48 project_root if isinstance(project_root, Path) else Path(project_root) 

49 ) 

50 from ...utils.runtime_paths import _sync_bin_path_from_root 

51 

52 _sync_bin_path_from_root(self._project_root / "bin") 

53 

54 def load_config( 

55 self, config_name: str, force_reload: bool = False 

56 ) -> dict[str, Any]: 

57 """Load a named configuration from the first matching path. 

58 

59 Parameters 

60 ---------- 

61 config_name: 

62 Configuration file name (without extension, e.g. ``"app"``). 

63 force_reload: 

64 Bypass cache and reload from disk. 

65 

66 Returns 

67 ------- 

68 dict[str, Any] 

69 Loaded configuration data, or empty dict on failure. 

70 """ 

71 if not force_reload and config_name in self._config_cache: 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true

72 return self._config_cache[config_name] 

73 

74 config_file = self._resolve_config_file(config_name) 

75 

76 if not config_file: 

77 warn_user( 

78 code="config.service.missing_file", 

79 user_message=f"No configuration file found for '{config_name}'", 

80 log_message=f"No configuration file found for '{config_name}'", 

81 ) 

82 get_printer().verbose_msg( 

83 f"Searched paths: {self.get_config_paths(config_name)}" 

84 ) 

85 return {} 

86 

87 try: 

88 with open(config_file, encoding="utf-8") as f: 

89 config_data = yaml.safe_load(f) 

90 

91 self._config_cache[config_name] = config_data 

92 self._config_files[config_name] = config_file 

93 

94 get_printer().verbose_msg( 

95 f"Configuration '{config_name}' loaded from: {config_file}" 

96 ) 

97 return config_data 

98 

99 except Exception as e: 

100 get_printer().error(f"Error loading '{config_name}': {e}") 

101 return {} 

102 

103 def get_config_value( 

104 self, config_name: str, key_path: str, default: Any = None 

105 ) -> Any: 

106 """Read a specific value from a configuration using dot-notation key path. 

107 

108 Parameters 

109 ---------- 

110 config_name: 

111 Configuration file name. 

112 key_path: 

113 Dot-separated path (e.g. ``"app.name"`` or ``"palette.dark"``). 

114 default: 

115 Value returned when key is absent. 

116 """ 

117 config = self.load_config(config_name) 

118 keys = key_path.split(".") 

119 current = config 

120 

121 for key in keys: 

122 if isinstance(current, dict) and key in current: 122 ↛ 125line 122 didn't jump to line 125 because the condition on line 122 was always true

123 current = current[key] 

124 else: 

125 return default 

126 

127 return current 

128 

129 def save_config(self, config_name: str, config_data: dict[str, Any]) -> bool: 

130 """Persist a named configuration to the project directory. 

131 

132 Parameters 

133 ---------- 

134 config_name: 

135 Configuration file name. 

136 config_data: 

137 Data to serialise as YAML. 

138 

139 Returns 

140 ------- 

141 bool 

142 ``True`` if the write succeeded. 

143 """ 

144 config_file: Path 

145 if config_name in self._config_files: 

146 # Keep writes consistent with the file originally loaded. 

147 config_file = self._config_files[config_name] 

148 config_file.parent.mkdir(parents=True, exist_ok=True) 

149 else: 

150 config_dir = get_bin_path() / "config" 

151 config_dir.mkdir(parents=True, exist_ok=True) 

152 config_file = config_dir / f"{config_name}.config.yaml" 

153 

154 try: 

155 with open(config_file, "w", encoding="utf-8") as f: 

156 yaml.dump(config_data, f, default_flow_style=False, allow_unicode=True) 

157 

158 self._config_cache[config_name] = config_data 

159 self._config_files[config_name] = config_file 

160 

161 get_printer().verbose_msg( 

162 f"Configuration '{config_name}' saved: {config_file}" 

163 ) 

164 return True 

165 

166 except Exception as e: 

167 get_printer().error(f"Error saving '{config_name}': {e}") 

168 return False 

169 

170 # ----------------------------------------------------------- 

171 # Implementation-specific methods 

172 # ----------------------------------------------------------- 

173 

174 def get_config_paths(self, config_name: str) -> list[Path]: 

175 """Return candidate paths for *config_name* in priority order. 

176 

177 Parameters 

178 ---------- 

179 config_name: 

180 Logical configuration name (e.g. ``"app"``, ``"languages"``, 

181 ``"theme"``). Files are resolved as ``<name>.config.yaml``. 

182 """ 

183 config_file = f"{config_name}.config.yaml" 

184 paths: list[Path] = [] 

185 

186 # 1. Configured bin directory 

187 paths.append(get_bin_path() / "config" / config_file) 

188 

189 # 3. Package resources — relative to cwd 

190 paths.append(Path.cwd() / "ezqt_app" / "resources" / "config" / config_file) 

191 

192 # 4. Package resources — relative to APP_PATH 

193 paths.append(APP_PATH / "resources" / "config" / config_file) 

194 

195 return paths 

196 

197 def copy_package_configs_to_project(self) -> bool: 

198 """Copy package configuration files into the child project. 

199 

200 Returns 

201 ------- 

202 bool 

203 ``True`` if the operation succeeded. 

204 """ 

205 if not self._project_root: 

206 get_printer().error("[ConfigService] No project root defined") 

207 return False 

208 

209 package_config_dir = self._find_package_config_dir() 

210 

211 if not package_config_dir: 

212 get_printer().error( 

213 f"[ConfigService] EzQt_App package not found. Tested paths: " 

214 f"{[str(p) for p in [Path.cwd(), APP_PATH]]}" 

215 ) 

216 return False 

217 

218 project_config_dir = get_bin_path() / "config" 

219 project_config_dir.mkdir(parents=True, exist_ok=True) 

220 

221 copied_files: list[str] = [] 

222 

223 try: 

224 for config_file in package_config_dir.glob("*.yaml"): 

225 target_file = project_config_dir / config_file.name 

226 

227 if target_file.exists(): 

228 get_printer().verbose_msg(f"Existing file, ignored: {target_file}") 

229 continue 

230 

231 shutil.copy2(config_file, target_file) 

232 copied_files.append(config_file.name) 

233 get_printer().action( 

234 f"[ConfigService] Configuration copied: {config_file.name}" 

235 ) 

236 

237 if copied_files: 

238 get_printer().action( 

239 "[ConfigService] " 

240 f"{len(copied_files)} configurations copied to project" 

241 ) 

242 

243 return True 

244 

245 except Exception as e: 

246 get_printer().error(f"Error copying configurations: {e}") 

247 return False 

248 

249 def clear_cache(self) -> None: 

250 """Invalidate the in-memory configuration cache.""" 

251 self._config_cache.clear() 

252 self._config_files.clear() 

253 get_printer().verbose_msg("Configuration cache cleared") 

254 

255 def get_loaded_configs(self) -> dict[str, Path]: 

256 """Return a snapshot of currently cached configuration paths.""" 

257 return self._config_files.copy() 

258 

259 # ----------------------------------------------------------- 

260 # Private helpers 

261 # ----------------------------------------------------------- 

262 

263 def _resolve_config_file(self, config_name: str) -> Path | None: 

264 """Return the first existing path for *config_name*, or ``None``.""" 

265 for path in self.get_config_paths(config_name): 

266 if path.exists(): 

267 return path 

268 return None 

269 

270 def _find_package_config_dir(self) -> Path | None: 

271 """Locate the ``resources/config`` directory inside the installed package.""" 

272 package_root = _get_installed_package_root() 

273 if package_root is not None: 

274 installed_config_dir = package_root / "resources" / "config" 

275 if installed_config_dir.exists(): 

276 return installed_config_dir 

277 

278 # 1. Walk up from cwd looking for an ezqt_app package dir 

279 current = Path.cwd() 

280 while current.parent != current: 

281 candidate = current / "ezqt_app" 

282 if candidate.exists() and (candidate / "resources" / "config").exists(): 

283 config_dir = candidate / "resources" / "config" 

284 get_printer().verbose_msg(f"EzQt_App package found: {candidate}") 

285 get_printer().verbose_msg(f"Configuration directory: {config_dir}") 

286 return config_dir 

287 current = current.parent 

288 

289 # 2. Search sys.path entries 

290 for path in sys.path: 

291 candidate = Path(path) / "ezqt_app" 

292 if candidate.exists() and (candidate / "resources" / "config").exists(): 

293 return candidate / "resources" / "config" 

294 

295 # 3. Fallback — legacy known paths 

296 for fallback in [ 

297 Path.cwd() / "ezqt_app" / "resources" / "config", 

298 APP_PATH / "resources" / "config", 

299 APP_PATH / "ezqt_app" / "resources" / "config", 

300 ]: 

301 if fallback.exists(): 

302 get_printer().verbose_msg( 

303 f"Configuration directory found (fallback): {fallback}" 

304 ) 

305 return fallback 

306 

307 return None 

308 

309 

310# /////////////////////////////////////////////////////////////// 

311# SINGLETONS / FUNCTIONS 

312# /////////////////////////////////////////////////////////////// 

313 

314 

315def _get_installed_package_root() -> Path | None: 

316 """Return installed ``ezqt_app`` package root when discoverable.""" 

317 try: 

318 import importlib.util 

319 

320 spec = importlib.util.find_spec("ezqt_app") 

321 if spec is not None: 321 ↛ 337line 321 didn't jump to line 337 because the condition on line 321 was always true

322 if spec.submodule_search_locations: 322 ↛ 328line 322 didn't jump to line 328 because the condition on line 322 was always true

323 location = next(iter(spec.submodule_search_locations), None) 

324 if location: 324 ↛ 328line 324 didn't jump to line 328 because the condition on line 324 was always true

325 candidate = Path(location) 

326 if candidate.exists(): 326 ↛ 328line 326 didn't jump to line 328 because the condition on line 326 was always true

327 return candidate 

328 if spec.origin: 

329 candidate = Path(spec.origin).resolve().parent 

330 if candidate.exists(): 

331 return candidate 

332 except Exception as e: 

333 get_printer().verbose_msg( 

334 f"Could not resolve installed package root for ezqt_app: {e}" 

335 ) 

336 

337 return None 

338 

339 

340def _resource_candidates(resource_path: str) -> list[Path]: 

341 """Build candidate paths for a package resource in priority order.""" 

342 rel_path = Path(resource_path.replace("\\", "/")) 

343 candidates: list[Path] = [] 

344 

345 package_root = _get_installed_package_root() 

346 if package_root is not None: 346 ↛ 349line 346 didn't jump to line 349 because the condition on line 346 was always true

347 candidates.append(package_root / rel_path) 

348 

349 candidates.extend( 

350 [ 

351 APP_PATH / "ezqt_app" / rel_path, 

352 APP_PATH / rel_path, 

353 Path.cwd() / "ezqt_app" / rel_path, 

354 ] 

355 ) 

356 return candidates 

357 

358 

359# /////////////////////////////////////////////////////////////// 

360# FUNCTIONS 

361# /////////////////////////////////////////////////////////////// 

362def get_config_service() -> ConfigService: 

363 """Return the singleton configuration service instance.""" 

364 from .._registry import ServiceRegistry 

365 

366 return ServiceRegistry.get(ConfigService, ConfigService) 

367 

368 

369def get_package_resource(resource_path: str) -> Path: 

370 """Return the filesystem path of an installed package resource. 

371 

372 Parameters 

373 ---------- 

374 resource_path: 

375 Resource path relative to the ``ezqt_app`` package root. 

376 

377 Returns 

378 ------- 

379 Path 

380 Resolved path to the resource. 

381 """ 

382 for candidate in _resource_candidates(resource_path): 382 ↛ 386line 382 didn't jump to line 386 because the loop on line 382 didn't complete

383 if candidate.exists(): 383 ↛ 382line 383 didn't jump to line 382 because the condition on line 383 was always true

384 return candidate 

385 

386 try: 

387 pkg_candidate = Path(str(_pkg_files("ezqt_app").joinpath(resource_path))) 

388 if pkg_candidate.exists(): 

389 return pkg_candidate 

390 except Exception as e: 

391 get_printer().verbose_msg( 

392 f"importlib.resources lookup failed for '{resource_path}': {e}" 

393 ) 

394 

395 package_root = _get_installed_package_root() 

396 if package_root is not None: 

397 return package_root / Path(resource_path.replace("\\", "/")) 

398 return APP_PATH / "ezqt_app" / Path(resource_path.replace("\\", "/")) 

399 

400 

401def get_package_resource_content(resource_path: str) -> str: 

402 """Return the decoded UTF-8 content of an installed package resource. 

403 

404 Parameters 

405 ---------- 

406 resource_path: 

407 Resource path relative to the ``ezqt_app`` package root. 

408 

409 Returns 

410 ------- 

411 str 

412 Resource file content. 

413 

414 Raises 

415 ------ 

416 FileNotFoundError 

417 If the resource cannot be located by any strategy. 

418 """ 

419 try: 

420 return ( 

421 _pkg_files("ezqt_app").joinpath(resource_path).read_text(encoding="utf-8") 

422 ) 

423 except Exception: 

424 file_path = get_package_resource(resource_path) 

425 if file_path.exists(): 

426 with open(file_path, encoding="utf-8") as f: 

427 return f.read() 

428 raise FileNotFoundError(f"Resource not found: {resource_path}") from None