from __future__ import annotations import copy import importlib import sys import traceback from dataclasses import dataclass, field from typing import Any from src.result_types import Result, ErrorInfo, ErrorKind @dataclass class HotModule: name: str file_path: str state_keys: list[str] = field(default_factory=list) delegation_targets: list[str] = field(default_factory=list) class HotReloader: HOT_MODULES: dict[str, HotModule] = {} last_error: str | None = None is_error_state: bool = False @classmethod def register(cls, module: HotModule) -> None: if module.name in cls.HOT_MODULES: raise ValueError(f"Module {module.name} already registered") cls.HOT_MODULES[module.name] = module @classmethod def capture_state(cls, app: Any, state_keys: list[str]) -> dict[str, Any]: return {key: copy.deepcopy(getattr(app, key, None)) for key in state_keys if hasattr(app, key)} @classmethod def restore_state(cls, app: Any, state: dict[str, Any]) -> None: for key, value in state.items(): setattr(app, key, value) @classmethod def reload(cls, module_name: str, app: Any) -> Result[bool]: if module_name not in cls.HOT_MODULES: err_msg = f"Module {module_name} not registered" cls.last_error = err_msg cls.is_error_state = True return Result(data=False, errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=err_msg, source=f"hot_reloader.reload[{module_name}]")]) hm = cls.HOT_MODULES[module_name] state = cls.capture_state(app, hm.state_keys) try: if module_name in sys.modules: old_module = sys.modules[module_name] importlib.reload(old_module) else: importlib.import_module(module_name) cls.last_error = None cls.is_error_state = False return Result(data=True) except Exception as e: cls.restore_state(app, state) tb = traceback.format_exc() cls.last_error = tb cls.is_error_state = True return Result(data=False, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source=f"hot_reloader.reload[{module_name}]", original=e)]) @classmethod def reload_all(cls, app: Any) -> Result[bool]: errors: list[ErrorInfo] = [] for name in cls.HOT_MODULES: result = cls.reload(name, app) if not result.ok: errors.extend(result.errors) return Result(data=len(errors) == 0, errors=errors)