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, 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: watchdog.observers.Observer 

75 """The ``watchdog`` observer, created by ``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.dirEntries = {} 

85 self.filePaths = set() 

86 self.excludePatterns = [] 

87 

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

89 """Add a directory to watch. 

90 

91 Args: 

92 dirname: Directory path. 

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

94 recursive: Also watch subdirectories. 

95 """ 

96 d = str(dirname) 

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

98 

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

100 """Add a file to watch. 

101 

102 Args: 

103 filename: File path. 

104 """ 

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

106 

107 def exclude(self, path_pattern: str): 

108 """Exclude paths from watching. 

109 

110 Args: 

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

112 """ 

113 self.excludePatterns.append(path_pattern) 

114 

115 def start(self): 

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

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

118 

119 h = _Handler(self) 

120 

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

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

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

124 

125 for f in self.filePaths: 

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

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

128 

129 self.observer.start() 

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

131 

132 def stop(self): 

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

134 if self.observer is not None: 

135 self.observer.stop() 

136 self.observer.join() 

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

138 

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

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

141 

142 Args: 

143 ev: The event. 

144 """ 

145 if self.path_matches(ev.src_path): 

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

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

148 

149 def path_matches(self, path): 

150 """Check if a path is watched. 

151 

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

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

154 

155 Args: 

156 path: File path. 

157 

158 Returns: 

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

160 """ 

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

162 return False 

163 if path in self.filePaths: 

164 return True 

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

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

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

168 return True 

169 return False 

170 

171 

172class _Handler(watchdog.events.FileSystemEventHandler): 

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

174 

175 def __init__(self, obj: Watcher): 

176 super().__init__() 

177 self.obj = obj 

178 

179 def on_any_event(self, ev): 

180 if ev.event_type in _WATCH_EVENTS: 

181 self.obj.register(ev)