| 300 | |
| 301 | |
| 302 | class HookConfigManager: |
| 303 | def __init__( |
| 304 | self, |
| 305 | registry: AsyncHookRegistry, |
| 306 | settings_path: str | Path | None = None, |
| 307 | ) -> None: |
| 308 | self._registry = registry |
| 309 | self._settings_path = Path(settings_path) if settings_path else _get_settings_path() |
| 310 | self._snapshot: HookConfigSnapshot | None = None |
| 311 | self._last_mtime: float = 0.0 |
| 312 | |
| 313 | @property |
| 314 | def snapshot(self) -> HookConfigSnapshot | None: |
| 315 | return self._snapshot |
| 316 | |
| 317 | async def load(self) -> HookConfigSnapshot: |
| 318 | snapshot = load_hooks_from_settings(self._settings_path) |
| 319 | self._snapshot = snapshot |
| 320 | |
| 321 | await self._registry.clear_source(HookSource.USER_SETTINGS) |
| 322 | |
| 323 | for event_name, hook_configs in snapshot.hooks.items(): |
| 324 | for config in hook_configs: |
| 325 | if event_name in ALL_HOOK_EVENTS: |
| 326 | await self._registry.register( |
| 327 | event_name, # type: ignore[arg-type] |
| 328 | config, |
| 329 | HookSource.USER_SETTINGS, |
| 330 | ) |
| 331 | |
| 332 | try: |
| 333 | self._last_mtime = self._settings_path.stat().st_mtime |
| 334 | except OSError: |
| 335 | self._last_mtime = 0.0 |
| 336 | |
| 337 | return snapshot |
| 338 | |
| 339 | async def reload_if_changed(self) -> bool: |
| 340 | try: |
| 341 | current_mtime = self._settings_path.stat().st_mtime |
| 342 | except OSError: |
| 343 | return False |
| 344 | |
| 345 | if current_mtime > self._last_mtime: |
| 346 | await self.load() |
| 347 | return True |
| 348 | |
| 349 | return False |
| 350 | |
| 351 | async def validate(self) -> list[HookValidationError]: |
| 352 | if not self._settings_path.exists(): |
| 353 | return [] |
| 354 | |
| 355 | try: |
| 356 | data = json.loads(self._settings_path.read_text(encoding="utf-8")) |
| 357 | except (json.JSONDecodeError, OSError): |
| 358 | return [HookValidationError( |
| 359 | event="", |
no outgoing calls