Start watching *raw_dir* and block until Ctrl+C. Args: raw_dir: Directory to watch for file changes. callback: Called with sorted list of new/modified file paths. debounce: Debounce delay in seconds. Defaults to 2.0.
(
raw_dir: Path,
callback: Callable[[list[str]], None],
debounce: float = 2.0,
)
| 78 | |
| 79 | |
| 80 | def watch_directory( |
| 81 | raw_dir: Path, |
| 82 | callback: Callable[[list[str]], None], |
| 83 | debounce: float = 2.0, |
| 84 | ) -> None: |
| 85 | """Start watching *raw_dir* and block until Ctrl+C. |
| 86 | |
| 87 | Args: |
| 88 | raw_dir: Directory to watch for file changes. |
| 89 | callback: Called with sorted list of new/modified file paths. |
| 90 | debounce: Debounce delay in seconds. Defaults to 2.0. |
| 91 | """ |
| 92 | handler = DebouncedHandler(callback, debounce_seconds=debounce) |
| 93 | observer = Observer() |
| 94 | observer.schedule(handler, str(raw_dir), recursive=True) |
| 95 | observer.start() |
| 96 | try: |
| 97 | while observer.is_alive(): |
| 98 | observer.join(timeout=1.0) |
| 99 | except KeyboardInterrupt: |
| 100 | observer.stop() |
| 101 | observer.join() |
no test coverage detected