Coverage for src / rendercv / cli / render_command / watcher.py: 100%

1import contextlib (empty)

2import pathlib (empty)

3from collections.abc import Callable (empty)

4 

5import typer (empty)

6import watchdog.events (empty)

7import watchdog.observers (empty)

8 

9 

10class EventHandler(watchdog.events.FileSystemEventHandler): (empty)

11 """Trigger a callback when a watched file is modified. 

12 

13 Args: 

14 function: Callback to invoke on modification. 

15 watched_files: Absolute path strings of files to monitor. 

16 """ 

17 

18 def __init__(self, function: Callable[[], None], watched_files: set[str]) -> None: (empty)

19 super().__init__() (empty)

20 self.function = function (empty)

21 self.watched_files = watched_files (empty)

22 

23 def on_modified( (empty)

24 self, 

25 event: watchdog.events.DirModifiedEvent | watchdog.events.FileModifiedEvent, 

26 ) -> None: 

27 if event.src_path not in self.watched_files: (empty)

28 return (empty)

29 with contextlib.suppress(typer.Exit): (empty)

30 self.function() (empty)

31 

32 

33def run_function_if_files_change( (empty)

34 file_paths: list[pathlib.Path], 

35 function: Callable[[], None], 

36) -> None: 

37 """Watch files and re-run function when any is modified. 

38 

39 Why: 

40 Watch mode lets users edit CV YAML and see results instantly. 

41 All config files (main input, design, locale, settings) must be 

42 monitored so edits to any trigger a re-render. 

43 

44 Args: 

45 file_paths: File paths to watch. 

46 function: Zero-argument callback to invoke on file change. 

47 """ 

48 watched_files = {str(fp.absolute()) for fp in file_paths} (empty)

49 

50 # Watch parent directories (file-level watching is unreliable across platforms) 

51 dirs_to_schedule = {str(fp.absolute().parent) for fp in file_paths} (empty)

52 

53 event_handler = EventHandler(function, watched_files) (empty)

54 

55 observer = watchdog.observers.Observer() (empty)

56 for directory in dirs_to_schedule: (empty)

57 observer.schedule(event_handler, directory, recursive=False) (empty)

58 observer.start() (empty)

59 

60 # Run immediately for the first render: 

61 with contextlib.suppress(typer.Exit): (empty)

62 function() (empty)

63 

64 try: (empty)

65 observer.join() (empty)

66 except KeyboardInterrupt: (empty)

67 observer.stop() (empty)

68 observer.join() (empty)