Coverage for gws-app/gws/lib/watcher/__init__.py: 100%

1"""File system watcher. 

2 

3Monitors file system events (creation, deletion, modification and movement) using ``watchdog``, 

4and calls a notification callback when an event occurs on a watched path. 

5 

6A ``Watcher`` can watch whole directories, optionally recursively and restricted to file names 

7matching a regular expression, and individual files. Paths matching an exclude pattern are ignored. 

8The callback is called from the observer thread, not from the thread that started the watcher. 

9 

10Example:: 

11 

12 import gws.lib.watcher 

13 

14 def on_change(event_type, path): 

15 print(event_type, path) 

16 

17 w = gws.lib.watcher.new(on_change) 

18 w.add_directory('/data/config', file_pattern=r'[.]json$', recursive=True) 

19 w.add_file('/data/config.cx') 

20 w.exclude(r'/[.]git/') 

21 w.start() 

22 ... 

23 w.stop() 

24""" 

25 

26from typing import Callable, Optional, TypeAlias 

27import os 

28import re 

29 

30import watchdog.events 

31import watchdog.observers 

32 

33import gws 

34 

35_WATCH_EVENTS = { 

36 watchdog.events.EVENT_TYPE_MOVED, 

37 watchdog.events.EVENT_TYPE_DELETED, 

38 watchdog.events.EVENT_TYPE_CREATED, 

39 watchdog.events.EVENT_TYPE_MODIFIED, 

40} 

41 

42_EVENTS = [] 

43 

44 

45class _DirEntry: 

46 """A watched directory.""" 

47 

48 def __init__(self, dirname, pattern, recursive): 

49 self.dirname = dirname 

50 self.pattern = pattern 

51 self.recursive = recursive 

52 

53 

54_NotifyFn: TypeAlias = Callable[[str, str], None] 

55 

56 

57def new(notify: _NotifyFn): 

58 """Create a new watcher. 

59 

60 Args: 

61 notify: A callback function that will be called when an event occurs. 

62 It accepts two arguments: the event type and the path of the file. 

63 The callback is called from a different thread than the one that created the watcher. 

64 

65 Returns: 

66 The watcher, not started yet. 

67 """ 

68 return Watcher(notify) 

69 

70 

71class Watcher: 

72 """File system watcher.""" 

73 

74 observer: Optional[watchdog.observers.Observer] 

75 """The ``watchdog`` observer, created by ``start``. ``None`` before ``start``.""" 

76 

77 def __init__(self, notify: _NotifyFn): 

78 """Create a watcher. 

79 

80 Args: 

81 notify: Callback, called with the event type and the path. 

82 """ 

83 self.notify = notify 

84 self.observer = None 

85 self.dirEntries = {} 

86 self.filePaths = set() 

87 self.excludePatterns = [] 

88 

89 def add_directory(self, dirname: str | os.PathLike, file_pattern: str = '', recursive: bool = False): 

90 """Add a directory to watch. 

91 

92 Args: 

93 dirname: Directory path. 

94 file_pattern: Regular expression to search for in file names. If empty, all files match. 

95 recursive: Also watch subdirectories. 

96 """ 

97 d = str(dirname) 

98 self.dirEntries[d] = _DirEntry(d, file_pattern or '.', recursive) 

99 

100 def add_file(self, filename: str | os.PathLike): 

101 """Add a file to watch. 

102 

103 Args: 

104 filename: File path. 

105 """ 

106 self.filePaths.add(str(filename)) 

107 

108 def exclude(self, path_pattern: str): 

109 """Exclude paths from watching. 

110 

111 Args: 

112 path_pattern: Regular expression to search for in the full path. 

113 """ 

114 self.excludePatterns.append(path_pattern) 

115 

116 def start(self): 

117 """Start watching the added directories and files in a background thread.""" 

118 self.observer = watchdog.observers.Observer() 

119 

120 h = _Handler(self) 

121 

122 for de in self.dirEntries.values(): 

123 gws.log.debug(f'watcher: watching {de.dirname!r}') 

124 self.observer.schedule(h, de.dirname, recursive=de.recursive) 

125 

126 for f in self.filePaths: 

127 gws.log.debug(f'watcher: watching {f!r}') 

128 self.observer.schedule(h, os.path.dirname(f), recursive=False) 

129 

130 self.observer.start() 

131 gws.log.debug(f'watcher: started with {self.observer.__class__.__name__}') 

132 

133 def stop(self): 

134 """Stop watching and wait for the observer thread to finish.""" 

135 if self.observer is not None: 

136 self.observer.stop() 

137 self.observer.join() 

138 gws.log.debug(f'watcher: stopped') 

139 

140 def register(self, ev: watchdog.events.FileSystemEvent): 

141 """Handle a file system event, calling the callback if the path is watched. 

142 

143 Args: 

144 ev: The event. 

145 """ 

146 if self.path_matches(ev.src_path): 

147 gws.log.debug(f'watcher: {ev.event_type} {ev.src_path}') 

148 self.notify(ev.event_type, ev.src_path) 

149 

150 def path_matches(self, path): 

151 """Check if a path is watched. 

152 

153 A path is watched if it is not excluded, and is either an added file 

154 or a file in an added directory whose name matches the directory's file pattern. 

155 

156 Args: 

157 path: File path. 

158 

159 Returns: 

160 ``True`` if the path is watched. 

161 """ 

162 if any(re.search(ex, path) for ex in self.excludePatterns): 

163 return False 

164 if path in self.filePaths: 

165 return True 

166 d, f = os.path.split(path) 

167 for de in self.dirEntries.values(): 

168 if (d == de.dirname or (de.recursive and d.startswith(de.dirname + '/'))) and re.search(de.pattern, f): 

169 return True 

170 return False 

171 

172 

173class _Handler(watchdog.events.FileSystemEventHandler): 

174 """Event handler that passes watched event types to the watcher.""" 

175 

176 def __init__(self, obj: Watcher): 

177 super().__init__() 

178 self.obj = obj 

179 

180 def on_any_event(self, ev): 

181 if ev.event_type in _WATCH_EVENTS: 

182 self.obj.register(ev)