Class CodeWatcher

src/codegraphcontext/core/watcher.py:384–457

class CodeWatcher:
    def __init__(
        self,
        graph_builder: "GraphBuilder",
        job_manager="JobManager",
        use_polling: typing.Optional[bool] = None,
    ):
        self.graph_builder = graph_builder
        observer_cls = PollingObserver if should_use_polling_observer(use_polling) else Observer
        self.observer = observer_cls()

        self.watched_paths = set()
        self.watches = {}
        self.handlers = {}

    def watch_directory(
        self,
        path: str,
        perform_initial_scan: bool = True,
        cgcignore_path: str = None,
        sync_on_start: bool = False,
    ):
        path_obj = Path(path).resolve()
        path_str = str(path_obj)

        if path_str in self.watched_paths:
            return {"message": "Already watching"}

        handler = RepositoryEventHandler(
            self.graph_builder,
            path_obj,
            perform_initial_scan=perform_initial_scan,
            sync_on_start=sync_on_start,
            cgcignore_path=cgcignore_path,
        )

        watch = self.observer.schedule(handler, path_str, recursive=True)

        self.watches[path_str] = watch
        self.handlers[path_str] = handler
        self.watched_paths.add(path_str)

        return {"message": f"Watching {path_str}"}

    def unwatch_directory(self, path: str):
        path_str = str(Path(path).resolve())

        handler = self.handlers.pop(path_str, None)
        if handler:
            handler.cancel_timers()

        watch = self.watches.pop(path_str, None)
        if watch:
            self.observer.unschedule(watch)

        self.watched_paths.discard(path_str)

        return {"message": f"Stopped watching {path_str}"}