Skip to content

watcher

dev_tool.services.watcher

__all__ = ['FileWatcherService'] module-attribute

FileWatcherService

A service class for watching file changes.

This class provides methods for monitoring files for changes and triggering callbacks.

The constructor for the FileWatcherService class.

Parameters:

  • interval (float, default: 1.0 ) –

    The polling interval in seconds.

Source code in dev_tool/services/watcher/service.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(self, interval: float = 1.0) -> None:
    """
    The constructor for the FileWatcherService class.

    :param interval: The polling interval in seconds.
    """

    super().__init__()

    self.interval = interval
    self._watched = {}
    self._callback = {}
    self._running = False
    self._thread: threading.Thread | None = None
    self._stop_event = threading.Event()

interval = interval instance-attribute

watch

A method that adds a file to the watch list.

Parameters:

  • file (Path) –

    The path to the file to watch.

  • callback (Callable[[Path], None]) –

    The callback function to call when the file changes.

Source code in dev_tool/services/watcher/service.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def watch(self, file: Path, callback: Callable[[Path], None]) -> None:
    """
    A method that adds a file to the watch list.

    :param file: The path to the file to watch.
    :param callback: The callback function to call when the file changes.
    """

    try:
        path = file.absolute()

        if path not in self._watched:
            self._watched[path] = {
                'last_modified': self._get_last_modified(path),
                'exists': path.exists()
            }

            self._callback[path] = []

        if callback not in self._callback[path]:
            self._callback[path].append(callback)
    except Exception:
        message = f'Failed to watch file: {file}'
        log.exception(message)

        raise FileWatchingError(message) from None

unwatch

A method that removes a file from the watch list.

Parameters:

  • file (Path) –

    The path to the file to unwatch.

  • callback (Callable[[Path], None] | None, default: None ) –

    The specific callback to remove, or None to remove all callbacks.

Source code in dev_tool/services/watcher/service.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def unwatch(self, file: Path, callback: Callable[[Path], None] | None = None) -> None:
    """
    A method that removes a file from the watch list.

    :param file: The path to the file to unwatch.
    :param callback: The specific callback to remove, or None to remove all callbacks.
    """

    try:
        path = file.absolute()

        if path not in self._watched:
            return

        if callback is None:
            self._callback.pop(path, None)
            self._watched.pop(path, None)

            return

        if path in self._callback and callback in self._callback[path]:
            self._callback[path].remove(callback)

            if not self._callback[path]:
                self._callback.pop(path, None)
                self._watched.pop(path, None)
    except Exception:
        message = f'Failed to unwatch file: {file}'
        log.exception(message)

        raise FileWatchingError(message) from None

start

A method that starts the file watching process.

Source code in dev_tool/services/watcher/service.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def start(self) -> None:
    """A method that starts the file watching process."""

    if self._running:
        return

    try:
        self._running = True
        self._stop_event.clear()
        self._thread = threading.Thread(target=self._poll, daemon=True)
        self._thread.start()
    except Exception:
        message = 'Failed to start file watcher thread'
        log.exception(message)

        raise ThreadingError(message) from None

stop

A method that stops the file watching process.

Source code in dev_tool/services/watcher/service.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def stop(self) -> None:
    """A method that stops the file watching process."""

    if not self._running:
        return

    try:
        self._running = False
        self._stop_event.set()

        if self._thread and self._thread.is_alive():
            self._thread.join(timeout=2.0)
    except Exception:
        message = 'Failed to stop file watcher thread'
        log.exception(message)

        raise ThreadingError(message) from None