Coverage for src / ezqt_app / services / config / config_service.py: 62.66%
299 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-06 13:12 +0000
1# ///////////////////////////////////////////////////////////////
2# SERVICES.CONFIG.CONFIG_SERVICE - Config service implementation
3# Project: ezqt_app
4# ///////////////////////////////////////////////////////////////
6"""Configuration service — full implementation (absorbs ConfigManager logic)."""
8from __future__ import annotations
10# ///////////////////////////////////////////////////////////////
11# IMPORTS
12# ///////////////////////////////////////////////////////////////
13# Standard library imports
14import shutil
15import sys
16from collections.abc import Mapping, MutableMapping
17from importlib.resources import files as _pkg_files
18from pathlib import Path
19from typing import Any
21import yaml
22from pydantic import BaseModel, ConfigDict, ValidationError
24# Third-party imports
25from ruamel.yaml import YAML
26from ruamel.yaml.comments import CommentedMap
28# Local imports
29from ...domain.ports.config_service import ConfigServiceProtocol
30from ...utils.diagnostics import warn_user
31from ...utils.printer import get_printer
32from ...utils.runtime_paths import APP_PATH, get_bin_path
35# ///////////////////////////////////////////////////////////////
36# PYDANTIC VALIDATION MODELS
37# ///////////////////////////////////////////////////////////////
38class _AppSectionSchema(BaseModel):
39 """Validation schema for the ``app`` section in app config."""
41 model_config = ConfigDict(extra="forbid")
43 name: str | None = None
44 description: str | None = None
45 app_width: int | None = None
46 app_min_width: int | None = None
47 app_height: int | None = None
48 app_min_height: int | None = None
49 debug_printer: bool | None = None
50 menu_panel_shrinked_width: int | None = None
51 menu_panel_extended_width: int | None = None
52 settings_panel_width: int | None = None
53 time_animation: int | None = None
54 settings_storage_root: str | None = None
55 config_version: int | None = None
58class _SettingsPanelOptionSchema(BaseModel):
59 """Validation schema for one settings panel entry."""
61 model_config = ConfigDict(extra="forbid")
63 type: str | None = None
64 label: str | None = None
65 default: Any = None
66 description: str | None = None
67 enabled: bool | None = None
68 options: list[str] | None = None
69 min: int | None = None
70 max: int | None = None
71 unit: str | None = None
74class _AppConfigSchema(BaseModel):
75 """Validation schema for ``app.config.yaml`` payload."""
77 model_config = ConfigDict(extra="forbid")
79 app: _AppSectionSchema | None = None
80 settings_panel: dict[str, _SettingsPanelOptionSchema] | None = None
83class _TranslationSectionSchema(BaseModel):
84 """Validation schema for the ``translation`` section."""
86 model_config = ConfigDict(extra="forbid")
88 collect_strings: bool | None = None
89 auto_translation_enabled: bool | None = None
90 auto_translate_new_strings: bool | None = None
91 save_to_ts_files: bool | None = None
92 cache_translations: bool | None = None
93 max_cache_age_days: int | None = None
96class _LanguageDetectionSectionSchema(BaseModel):
97 """Validation schema for the ``language_detection`` section."""
99 model_config = ConfigDict(extra="forbid")
101 auto_detect_language: bool | None = None
102 confidence_threshold: float | None = None
105class _SupportedLanguageSchema(BaseModel):
106 """Validation schema for one supported language entry."""
108 model_config = ConfigDict(extra="forbid")
110 code: str
111 name: str
112 native_name: str
113 description: str
116class _TranslationConfigSchema(BaseModel):
117 """Validation schema for ``translation.config.yaml`` payload."""
119 model_config = ConfigDict(extra="forbid")
121 translation: _TranslationSectionSchema | None = None
122 language_detection: _LanguageDetectionSectionSchema | None = None
123 supported_languages: list[_SupportedLanguageSchema] | None = None
126class _ThemeConfigSchema(BaseModel):
127 """Validation schema for ``theme.config.yaml`` payload."""
129 model_config = ConfigDict(extra="forbid")
131 palette: dict[str, Any]
134# ///////////////////////////////////////////////////////////////
135# CLASSES
136# ///////////////////////////////////////////////////////////////
137class ConfigService(ConfigServiceProtocol):
138 """Modular configuration service for EzQt_App."""
140 def __init__(self):
141 self._config_cache: dict[str, Any] = {}
142 self._config_files: dict[str, Path] = {}
143 self._project_root: Path | None = None
144 self._yaml_writer = YAML(typ="rt")
145 self._yaml_writer.preserve_quotes = True
146 self._yaml_writer.default_flow_style = False
147 self._yaml_writer.indent(mapping=2, sequence=4, offset=2)
149 # -----------------------------------------------------------
150 # Port methods (ConfigServiceProtocol)
151 # -----------------------------------------------------------
153 def set_project_root(self, project_root: Path | str) -> None:
154 """Set the active project root directory."""
155 self._project_root = (
156 project_root if isinstance(project_root, Path) else Path(project_root)
157 )
158 from ...utils.runtime_paths import _sync_bin_path_from_root
160 _sync_bin_path_from_root(self._project_root / "bin")
162 def load_config(
163 self, config_name: str, force_reload: bool = False
164 ) -> dict[str, Any]:
165 """Load a named configuration from the first matching path.
167 Parameters
168 ----------
169 config_name:
170 Configuration file name (without extension, e.g. ``"app"``).
171 force_reload:
172 Bypass cache and reload from disk.
174 Returns
175 -------
176 dict[str, Any]
177 Loaded configuration data, or empty dict on failure.
178 """
179 if not force_reload and config_name in self._config_cache: 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true
180 return self._config_cache[config_name]
182 config_file = self._resolve_config_file(config_name)
184 if not config_file:
185 warn_user(
186 code="config.service.missing_file",
187 user_message=f"No configuration file found for '{config_name}'",
188 log_message=f"No configuration file found for '{config_name}'",
189 )
190 get_printer().verbose_msg(
191 f"Searched paths: {self.get_config_paths(config_name)}"
192 )
193 return {}
195 try:
196 with open(config_file, encoding="utf-8") as f:
197 config_data = yaml.safe_load(f)
199 self._config_cache[config_name] = config_data
200 self._config_files[config_name] = config_file
202 get_printer().verbose_msg(
203 f"Configuration '{config_name}' loaded from: {config_file}"
204 )
205 return config_data
207 except Exception as e:
208 get_printer().error(f"Error loading '{config_name}': {e}")
209 return {}
211 def get_config_value(
212 self, config_name: str, key_path: str, default: Any = None
213 ) -> Any:
214 """Read a specific value from a configuration using dot-notation key path.
216 Parameters
217 ----------
218 config_name:
219 Configuration file name.
220 key_path:
221 Dot-separated path (e.g. ``"app.name"`` or ``"palette.dark"``).
222 default:
223 Value returned when key is absent.
224 """
225 config = self.load_config(config_name)
226 keys = key_path.split(".")
227 current = config
229 for key in keys:
230 if isinstance(current, dict) and key in current: 230 ↛ 233line 230 didn't jump to line 233 because the condition on line 230 was always true
231 current = current[key]
232 else:
233 return default
235 return current
237 def save_config(self, config_name: str, config_data: dict[str, Any]) -> bool:
238 """Persist a named configuration to the project directory.
240 Writes use ``ruamel.yaml`` round-trip mode to preserve existing comments,
241 key ordering, and formatting whenever the target file already exists.
242 Configuration reads stay on ``PyYAML`` for the current typed read path.
244 Parameters
245 ----------
246 config_name:
247 Configuration file name.
248 config_data:
249 Data to serialise as YAML.
251 Returns
252 -------
253 bool
254 ``True`` if the write succeeded.
255 """
256 config_file: Path
257 if config_name in self._config_files:
258 # Keep writes consistent with the file originally loaded.
259 config_file = self._config_files[config_name]
260 config_file.parent.mkdir(parents=True, exist_ok=True)
261 else:
262 config_dir = get_bin_path() / "config"
263 config_dir.mkdir(parents=True, exist_ok=True)
264 config_file = config_dir / f"{config_name}.config.yaml"
266 try:
267 if not self._validate_config_payload(config_name, config_data): 267 ↛ 268line 267 didn't jump to line 268 because the condition on line 267 was never true
268 return False
270 existing_doc: MutableMapping[str, Any] | None = None
271 if config_file.exists():
272 with open(config_file, encoding="utf-8") as f:
273 loaded_doc = self._yaml_writer.load(f)
274 if isinstance(loaded_doc, MutableMapping): 274 ↛ 277line 274 didn't jump to line 277 because the condition on line 274 was always true
275 existing_doc = loaded_doc
277 if existing_doc is None:
278 payload: MutableMapping[str, Any] = self._to_yaml_mapping(config_data)
279 else:
280 payload = self._merge_yaml_mapping(existing_doc, config_data)
282 with open(config_file, "w", encoding="utf-8") as f:
283 self._yaml_writer.dump(payload, f)
285 self._config_cache[config_name] = config_data
286 self._config_files[config_name] = config_file
288 get_printer().verbose_msg(
289 f"Configuration '{config_name}' saved: {config_file}"
290 )
291 return True
293 except Exception as e:
294 get_printer().error(f"Error saving '{config_name}': {e}")
295 return False
297 def _validate_config_payload(
298 self, config_name: str, config_data: dict[str, Any]
299 ) -> bool:
300 """Validate known configuration payloads with permissive Pydantic schemas."""
301 schema_map: dict[str, type[BaseModel]] = {
302 "app": _AppConfigSchema,
303 "translation": _TranslationConfigSchema,
304 "theme": _ThemeConfigSchema,
305 }
307 schema = schema_map.get(config_name)
308 if schema is None: 308 ↛ 309line 308 didn't jump to line 309 because the condition on line 308 was never true
309 return True
311 try:
312 schema.model_validate(config_data)
313 return True
314 except ValidationError as exc:
315 warn_user(
316 code="config.service.validation_failed",
317 user_message=(
318 f"Invalid '{config_name}' configuration; file was not written."
319 ),
320 log_message=(
321 f"Validation failed for '{config_name}' configuration: {exc}"
322 ),
323 )
324 get_printer().error(
325 f"Validation failed for '{config_name}' configuration: {exc}"
326 )
327 return False
329 def _to_yaml_mapping(self, value: Mapping[str, Any]) -> CommentedMap:
330 """Convert a standard mapping to a ruamel-compatible ``CommentedMap``."""
331 converted = CommentedMap()
332 for key, item in value.items():
333 converted[key] = self._to_yaml_value(item)
334 return converted
336 def _to_yaml_value(self, value: Any) -> Any:
337 """Recursively convert Python containers to ruamel round-trip containers."""
338 if isinstance(value, Mapping):
339 return self._to_yaml_mapping(value)
340 if isinstance(value, list): 340 ↛ 341line 340 didn't jump to line 341 because the condition on line 340 was never true
341 return [self._to_yaml_value(item) for item in value]
342 return value
344 def _merge_yaml_mapping(
345 self,
346 target: MutableMapping[str, Any],
347 source: Mapping[str, Any],
348 ) -> MutableMapping[str, Any]:
349 """Merge ``source`` values into ``target`` while preserving YAML metadata."""
350 for existing_key in list(target.keys()):
351 if existing_key not in source: 351 ↛ 352line 351 didn't jump to line 352 because the condition on line 351 was never true
352 del target[existing_key]
354 for key, source_value in source.items():
355 current_value = target.get(key)
356 if isinstance(source_value, Mapping) and isinstance(
357 current_value, MutableMapping
358 ):
359 target[key] = self._merge_yaml_mapping(current_value, source_value)
360 continue
362 if isinstance(source_value, Mapping): 362 ↛ 363line 362 didn't jump to line 363 because the condition on line 362 was never true
363 target[key] = self._to_yaml_mapping(source_value)
364 continue
366 if isinstance(source_value, list): 366 ↛ 367line 366 didn't jump to line 367 because the condition on line 366 was never true
367 target[key] = [self._to_yaml_value(item) for item in source_value]
368 continue
370 target[key] = source_value
372 return target
374 # -----------------------------------------------------------
375 # Implementation-specific methods
376 # -----------------------------------------------------------
378 def get_config_paths(self, config_name: str) -> list[Path]:
379 """Return candidate paths for *config_name* in priority order.
381 Parameters
382 ----------
383 config_name:
384 Logical configuration name (e.g. ``"app"``, ``"languages"``,
385 ``"theme"``). Files are resolved as ``<name>.config.yaml``.
386 """
387 config_file = f"{config_name}.config.yaml"
388 paths: list[Path] = []
390 # 1. Configured bin directory
391 paths.append(get_bin_path() / "config" / config_file)
393 # 3. Package resources — relative to cwd
394 paths.append(Path.cwd() / "ezqt_app" / "resources" / "config" / config_file)
396 # 4. Package resources — relative to APP_PATH
397 paths.append(APP_PATH / "resources" / "config" / config_file)
399 return paths
401 def copy_package_configs_to_project(self) -> bool:
402 """Copy package configuration files into the child project.
404 Returns
405 -------
406 bool
407 ``True`` if the operation succeeded.
408 """
409 if not self._project_root:
410 get_printer().error("[ConfigService] No project root defined")
411 return False
413 package_config_dir = self._find_package_config_dir()
415 if not package_config_dir:
416 get_printer().error(
417 f"[ConfigService] EzQt_App package not found. Tested paths: "
418 f"{[str(p) for p in [Path.cwd(), APP_PATH]]}"
419 )
420 return False
422 project_config_dir = get_bin_path() / "config"
423 project_config_dir.mkdir(parents=True, exist_ok=True)
425 copied_files: list[str] = []
427 try:
428 for config_file in package_config_dir.glob("*.yaml"):
429 target_file = project_config_dir / config_file.name
431 if target_file.exists():
432 get_printer().verbose_msg(f"Existing file, ignored: {target_file}")
433 continue
435 shutil.copy2(config_file, target_file)
436 copied_files.append(config_file.name)
437 get_printer().action(
438 f"[ConfigService] Configuration copied: {config_file.name}"
439 )
441 if copied_files:
442 get_printer().action(
443 "[ConfigService] "
444 f"{len(copied_files)} configurations copied to project"
445 )
447 return True
449 except Exception as e:
450 get_printer().error(f"Error copying configurations: {e}")
451 return False
453 def clear_cache(self) -> None:
454 """Invalidate the in-memory configuration cache."""
455 self._config_cache.clear()
456 self._config_files.clear()
457 get_printer().verbose_msg("Configuration cache cleared")
459 def get_loaded_configs(self) -> dict[str, Path]:
460 """Return a snapshot of currently cached configuration paths."""
461 return self._config_files.copy()
463 # -----------------------------------------------------------
464 # Private helpers
465 # -----------------------------------------------------------
467 def _resolve_config_file(self, config_name: str) -> Path | None:
468 """Return the first existing path for *config_name*, or ``None``."""
469 for path in self.get_config_paths(config_name):
470 if path.exists():
471 return path
472 return None
474 def _find_package_config_dir(self) -> Path | None:
475 """Locate the ``resources/config`` directory inside the installed package."""
476 package_root = _get_installed_package_root()
477 if package_root is not None:
478 installed_config_dir = package_root / "resources" / "config"
479 if installed_config_dir.exists():
480 return installed_config_dir
482 # 1. Walk up from cwd looking for an ezqt_app package dir
483 current = Path.cwd()
484 while current.parent != current:
485 candidate = current / "ezqt_app"
486 if candidate.exists() and (candidate / "resources" / "config").exists():
487 config_dir = candidate / "resources" / "config"
488 get_printer().verbose_msg(f"EzQt_App package found: {candidate}")
489 get_printer().verbose_msg(f"Configuration directory: {config_dir}")
490 return config_dir
491 current = current.parent
493 # 2. Search sys.path entries
494 for path in sys.path:
495 candidate = Path(path) / "ezqt_app"
496 if candidate.exists() and (candidate / "resources" / "config").exists():
497 return candidate / "resources" / "config"
499 # 3. Fallback — legacy known paths
500 for fallback in [
501 Path.cwd() / "ezqt_app" / "resources" / "config",
502 APP_PATH / "resources" / "config",
503 APP_PATH / "ezqt_app" / "resources" / "config",
504 ]:
505 if fallback.exists():
506 get_printer().verbose_msg(
507 f"Configuration directory found (fallback): {fallback}"
508 )
509 return fallback
511 return None
514# ///////////////////////////////////////////////////////////////
515# SINGLETONS / FUNCTIONS
516# ///////////////////////////////////////////////////////////////
519def _get_installed_package_root() -> Path | None:
520 """Return installed ``ezqt_app`` package root when discoverable."""
521 try:
522 import importlib.util
524 spec = importlib.util.find_spec("ezqt_app")
525 if spec is not None: 525 ↛ 541line 525 didn't jump to line 541 because the condition on line 525 was always true
526 if spec.submodule_search_locations: 526 ↛ 532line 526 didn't jump to line 532 because the condition on line 526 was always true
527 location = next(iter(spec.submodule_search_locations), None)
528 if location: 528 ↛ 532line 528 didn't jump to line 532 because the condition on line 528 was always true
529 candidate = Path(location)
530 if candidate.exists(): 530 ↛ 532line 530 didn't jump to line 532 because the condition on line 530 was always true
531 return candidate
532 if spec.origin:
533 candidate = Path(spec.origin).resolve().parent
534 if candidate.exists():
535 return candidate
536 except Exception as e:
537 get_printer().verbose_msg(
538 f"Could not resolve installed package root for ezqt_app: {e}"
539 )
541 return None
544def _resource_candidates(resource_path: str) -> list[Path]:
545 """Build candidate paths for a package resource in priority order."""
546 rel_path = Path(resource_path.replace("\\", "/"))
547 candidates: list[Path] = []
549 package_root = _get_installed_package_root()
550 if package_root is not None: 550 ↛ 553line 550 didn't jump to line 553 because the condition on line 550 was always true
551 candidates.append(package_root / rel_path)
553 candidates.extend(
554 [
555 APP_PATH / "ezqt_app" / rel_path,
556 APP_PATH / rel_path,
557 Path.cwd() / "ezqt_app" / rel_path,
558 ]
559 )
560 return candidates
563# ///////////////////////////////////////////////////////////////
564# FUNCTIONS
565# ///////////////////////////////////////////////////////////////
566def get_config_service() -> ConfigService:
567 """Return the singleton configuration service instance."""
568 from .._registry import ServiceRegistry
570 return ServiceRegistry.get(ConfigService, ConfigService)
573def get_package_resource(resource_path: str) -> Path:
574 """Return the filesystem path of an installed package resource.
576 Parameters
577 ----------
578 resource_path:
579 Resource path relative to the ``ezqt_app`` package root.
581 Returns
582 -------
583 Path
584 Resolved path to the resource.
585 """
586 for candidate in _resource_candidates(resource_path): 586 ↛ 590line 586 didn't jump to line 590 because the loop on line 586 didn't complete
587 if candidate.exists(): 587 ↛ 586line 587 didn't jump to line 586 because the condition on line 587 was always true
588 return candidate
590 try:
591 pkg_candidate = Path(str(_pkg_files("ezqt_app").joinpath(resource_path)))
592 if pkg_candidate.exists():
593 return pkg_candidate
594 except Exception as e:
595 get_printer().verbose_msg(
596 f"importlib.resources lookup failed for '{resource_path}': {e}"
597 )
599 package_root = _get_installed_package_root()
600 if package_root is not None:
601 return package_root / Path(resource_path.replace("\\", "/"))
602 return APP_PATH / "ezqt_app" / Path(resource_path.replace("\\", "/"))
605def get_package_resource_content(resource_path: str) -> str:
606 """Return the decoded UTF-8 content of an installed package resource.
608 Parameters
609 ----------
610 resource_path:
611 Resource path relative to the ``ezqt_app`` package root.
613 Returns
614 -------
615 str
616 Resource file content.
618 Raises
619 ------
620 FileNotFoundError
621 If the resource cannot be located by any strategy.
622 """
623 try:
624 return (
625 _pkg_files("ezqt_app").joinpath(resource_path).read_text(encoding="utf-8")
626 )
627 except Exception:
628 file_path = get_package_resource(resource_path)
629 if file_path.exists():
630 with open(file_path, encoding="utf-8") as f:
631 return f.read()
632 raise FileNotFoundError(f"Resource not found: {resource_path}") from None