Coverage for src/lexigram/web/server/reload.py: 23%

1"""Hot reload hook for development. 

2 

3Detects file changes and reloads routes/controllers. 

4""" 

5 

6from __future__ import annotations 

7 

8import asyncio 

9import os 

10from typing import TYPE_CHECKING, Any 

11 

12from lexigram.logging import get_logger 

13 

14if TYPE_CHECKING: 

15 from collections.abc import Callable 

16 

17logger = get_logger(__name__) 

18 

19 

20class HotReloadManager: 

21 """Manages hot reload for routes and controllers during development.""" 

22 

23 def __init__( 

24 self, 

25 watch_paths: list[str] | None = None, 

26 on_reload: Callable | None = None, 

27 ): 

28 self.watch_paths = watch_paths or [] 

29 self.on_reload = on_reload 

30 self._running = False 

31 self._file_mtimes: dict[str, float] = {} 

32 self._task: asyncio.Task | None = None 

33 

34 async def start(self) -> None: 

35 """Start watching for file changes.""" 

36 self._running = True 

37 self._task = asyncio.create_task(self._watch_loop()) 

38 

39 async def stop(self) -> None: 

40 """Stop watching for file changes.""" 

41 self._running = False 

42 if self._task is not None and not self._task.done(): 

43 self._task.cancel() 

44 self._task = None 

45 

46 async def _watch_loop(self) -> None: 

47 """Watch loop that checks for file changes.""" 

48 while self._running: 

49 await asyncio.sleep(1.0) # Check every second 

50 

51 for path in self.watch_paths: 

52 if os.path.isfile(path): 

53 await self._check_file(path) 

54 elif os.path.isdir(path): 

55 for root, _dirs, files in os.walk(path): 

56 for f in files: 

57 if f.endswith(".py"): 

58 await self._check_file(os.path.join(root, f)) 

59 

60 async def _check_file(self, filepath: str) -> None: 

61 """Check if a file has been modified.""" 

62 try: 

63 mtime = os.path.getmtime(filepath) 

64 

65 if filepath in self._file_mtimes: 

66 if mtime > self._file_mtimes[filepath]: 

67 # File was modified 

68 await self._trigger_reload(filepath) 

69 

70 self._file_mtimes[filepath] = mtime 

71 except OSError: 

72 pass 

73 

74 async def _trigger_reload(self, filepath: str) -> None: 

75 """Trigger a reload when a file changes.""" 

76 if self.on_reload: 

77 try: 

78 await self.on_reload(filepath) 

79 except (RuntimeError, OSError) as exc: 

80 logger.debug("reload_handler_failed", error=str(exc)) 

81 

82 def add_watch_path(self, path: str) -> None: 

83 """Add a path to watch.""" 

84 if path not in self.watch_paths: 

85 self.watch_paths.append(path) 

86 

87 

88def create_hot_reload_middleware( 

89 watch_paths: list[str], 

90 on_reload: Callable | None = None, 

91) -> type: 

92 """Create a hot reload middleware. 

93 

94 Usage: 

95 app.add_middleware( 

96 HotReloadMiddleware, 

97 watch_paths=["/path/to/controllers"], 

98 ) 

99 """ 

100 

101 class HotReloadMiddleware: 

102 def __init__(self, app: Any) -> None: 

103 self.app = app 

104 self.manager = HotReloadManager( 

105 watch_paths=watch_paths, 

106 on_reload=on_reload, 

107 ) 

108 

109 async def __call__(self, scope: Any, receive: Any, send: Any) -> None: 

110 if scope["type"] == "lifespan": 

111 # Add startup/shutdown handlers 

112 async def on_startup() -> None: 

113 await self.manager.start() 

114 

115 async def on_shutdown() -> None: 

116 await self.manager.stop() 

117 

118 # This is a simplified implementation 

119 # In practice, you'd integrate with the lifespan context 

120 await self.app(scope, receive, send) 

121 else: 

122 await self.app(scope, receive, send) 

123 

124 return HotReloadMiddleware