ed
00eaa460fd
hot_reloader.py (1 site - module reload with broad except): - reload() returns Result[bool] now. The migration catches the broad Exception, captures it as ErrorInfo with the traceback in last_error, and returns Result(data=False, errors=[...]). - reload_all() returns Result[bool]; aggregates per-module errors. - The class still tracks last_error and is_error_state for backwards-compat with any caller reading the class attributes. warmup.py (5 sites): - L139 (on_complete callback fire): was except ...: pass. Now logs to sys.stderr with the exception. - L215 (_record_success callback fire): same. - L249 (_record_failure callback fire): same. - L276 (_log_canary stderr.write): was except OSError: pass. Now logs the OSError itself. - L300 (_log_summary stderr.write): same. startup_profiler.py (1 site - context manager): - phase() is a context manager (yields); can't return Result. The except inside the finally block now logs the OSError. Tests updated for hot_reloader to check result.ok and result.data. Tests verified: - tests/test_hot_reloader.py (9 tests) PASS - tests/test_hot_reload_integration.py (13 tests) PASS - tests/test_warmup.py (10 tests) PASS - tests/test_warmup_canaries.py (18 tests) PASS
76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
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)
|