Spaces:
Running
Running
File size: 1,366 Bytes
4bdb245 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 |
from collections.abc import Callable
from pathlib import Path
from typing import Any
from watchdog.events import (
FileCreatedEvent,
FileModifiedEvent,
FileSystemEvent,
FileSystemEventHandler,
)
from watchdog.observers import Observer
class IngestWatcher:
def __init__(
self, watch_path: Path, on_file_changed: Callable[[Path], None]
) -> None:
self.watch_path = watch_path
self.on_file_changed = on_file_changed
class Handler(FileSystemEventHandler):
def on_modified(self, event: FileSystemEvent) -> None:
if isinstance(event, FileModifiedEvent):
on_file_changed(Path(event.src_path))
def on_created(self, event: FileSystemEvent) -> None:
if isinstance(event, FileCreatedEvent):
on_file_changed(Path(event.src_path))
event_handler = Handler()
observer: Any = Observer()
self._observer = observer
self._observer.schedule(event_handler, str(watch_path), recursive=True)
def start(self) -> None:
self._observer.start()
while self._observer.is_alive():
try:
self._observer.join(1)
except KeyboardInterrupt:
break
def stop(self) -> None:
self._observer.stop()
self._observer.join()
|