Spaces:
Sleeping
Sleeping
File size: 9,492 Bytes
96772eb |
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 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 72 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 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 |
""":module: watchdog.tricks
:synopsis: Utility event handlers.
:author: [email protected] (Yesudeep Mangalapilly)
:author: [email protected] (Mickaël Schoentgen)
Classes
-------
.. autoclass:: Trick
:members:
:show-inheritance:
.. autoclass:: LoggerTrick
:members:
:show-inheritance:
.. autoclass:: ShellCommandTrick
:members:
:show-inheritance:
.. autoclass:: AutoRestartTrick
:members:
:show-inheritance:
"""
from __future__ import annotations
import contextlib
import functools
import logging
import os
import signal
import subprocess
import threading
import time
from watchdog.events import EVENT_TYPE_CLOSED_NO_WRITE, EVENT_TYPE_OPENED, FileSystemEvent, PatternMatchingEventHandler
from watchdog.utils import echo, platform
from watchdog.utils.event_debouncer import EventDebouncer
from watchdog.utils.process_watcher import ProcessWatcher
logger = logging.getLogger(__name__)
echo_events = functools.partial(echo.echo, write=lambda msg: logger.info(msg))
class Trick(PatternMatchingEventHandler):
"""Your tricks should subclass this class."""
def __repr__(self) -> str:
return f"<{type(self).__name__}>"
@classmethod
def generate_yaml(cls) -> str:
return f"""- {cls.__module__}.{cls.__name__}:
args:
- argument1
- argument2
kwargs:
patterns:
- "*.py"
- "*.js"
ignore_patterns:
- "version.py"
ignore_directories: false
"""
class LoggerTrick(Trick):
"""A simple trick that does only logs events."""
@echo_events
def on_any_event(self, event: FileSystemEvent) -> None:
pass
class ShellCommandTrick(Trick):
"""Executes shell commands in response to matched events."""
def __init__(
self,
shell_command: str,
*,
patterns: list[str] | None = None,
ignore_patterns: list[str] | None = None,
ignore_directories: bool = False,
wait_for_process: bool = False,
drop_during_process: bool = False,
):
super().__init__(
patterns=patterns,
ignore_patterns=ignore_patterns,
ignore_directories=ignore_directories,
)
self.shell_command = shell_command
self.wait_for_process = wait_for_process
self.drop_during_process = drop_during_process
self.process: subprocess.Popen[bytes] | None = None
self._process_watchers: set[ProcessWatcher] = set()
def on_any_event(self, event: FileSystemEvent) -> None:
if event.event_type in {EVENT_TYPE_OPENED, EVENT_TYPE_CLOSED_NO_WRITE}:
# FIXME: see issue #949, and find a way to better handle that scenario
return
from string import Template
if self.drop_during_process and self.is_process_running():
return
object_type = "directory" if event.is_directory else "file"
context = {
"watch_src_path": event.src_path,
"watch_dest_path": "",
"watch_event_type": event.event_type,
"watch_object": object_type,
}
if self.shell_command is None:
if hasattr(event, "dest_path"):
context["dest_path"] = event.dest_path
command = 'echo "${watch_event_type} ${watch_object} from ${watch_src_path} to ${watch_dest_path}"'
else:
command = 'echo "${watch_event_type} ${watch_object} ${watch_src_path}"'
else:
if hasattr(event, "dest_path"):
context["watch_dest_path"] = event.dest_path
command = self.shell_command
command = Template(command).safe_substitute(**context)
self.process = subprocess.Popen(command, shell=True)
if self.wait_for_process:
self.process.wait()
else:
process_watcher = ProcessWatcher(self.process, None)
self._process_watchers.add(process_watcher)
process_watcher.process_termination_callback = functools.partial(
self._process_watchers.discard,
process_watcher,
)
process_watcher.start()
def is_process_running(self) -> bool:
return bool(self._process_watchers or (self.process is not None and self.process.poll() is None))
class AutoRestartTrick(Trick):
"""Starts a long-running subprocess and restarts it on matched events.
The command parameter is a list of command arguments, such as
`['bin/myserver', '-c', 'etc/myconfig.ini']`.
Call `start()` after creating the Trick. Call `stop()` when stopping
the process.
"""
def __init__(
self,
command: list[str],
*,
patterns: list[str] | None = None,
ignore_patterns: list[str] | None = None,
ignore_directories: bool = False,
stop_signal: signal.Signals | int = signal.SIGINT,
kill_after: int = 10,
debounce_interval_seconds: int = 0,
restart_on_command_exit: bool = True,
):
if kill_after < 0:
error = "kill_after must be non-negative."
raise ValueError(error)
if debounce_interval_seconds < 0:
error = "debounce_interval_seconds must be non-negative."
raise ValueError(error)
super().__init__(
patterns=patterns,
ignore_patterns=ignore_patterns,
ignore_directories=ignore_directories,
)
self.command = command
self.stop_signal = stop_signal.value if isinstance(stop_signal, signal.Signals) else stop_signal
self.kill_after = kill_after
self.debounce_interval_seconds = debounce_interval_seconds
self.restart_on_command_exit = restart_on_command_exit
self.process: subprocess.Popen[bytes] | None = None
self.process_watcher: ProcessWatcher | None = None
self.event_debouncer: EventDebouncer | None = None
self.restart_count = 0
self._is_process_stopping = False
self._is_trick_stopping = False
self._stopping_lock = threading.RLock()
def start(self) -> None:
if self.debounce_interval_seconds:
self.event_debouncer = EventDebouncer(
debounce_interval_seconds=self.debounce_interval_seconds,
events_callback=lambda events: self._restart_process(),
)
self.event_debouncer.start()
self._start_process()
def stop(self) -> None:
# Ensure the body of the function is only run once.
with self._stopping_lock:
if self._is_trick_stopping:
return
self._is_trick_stopping = True
process_watcher = self.process_watcher
if self.event_debouncer is not None:
self.event_debouncer.stop()
self._stop_process()
# Don't leak threads: Wait for background threads to stop.
if self.event_debouncer is not None:
self.event_debouncer.join()
if process_watcher is not None:
process_watcher.join()
def _start_process(self) -> None:
if self._is_trick_stopping:
return
# windows doesn't have setsid
self.process = subprocess.Popen(self.command, preexec_fn=getattr(os, "setsid", None))
if self.restart_on_command_exit:
self.process_watcher = ProcessWatcher(self.process, self._restart_process)
self.process_watcher.start()
def _stop_process(self) -> None:
# Ensure the body of the function is not run in parallel in different threads.
with self._stopping_lock:
if self._is_process_stopping:
return
self._is_process_stopping = True
try:
if self.process_watcher is not None:
self.process_watcher.stop()
self.process_watcher = None
if self.process is not None:
try:
kill_process(self.process.pid, self.stop_signal)
except OSError:
# Process is already gone
pass
else:
kill_time = time.time() + self.kill_after
while time.time() < kill_time:
if self.process.poll() is not None:
break
time.sleep(0.25)
else:
# Process is already gone
with contextlib.suppress(OSError):
kill_process(self.process.pid, 9)
self.process = None
finally:
self._is_process_stopping = False
@echo_events
def on_any_event(self, event: FileSystemEvent) -> None:
if event.event_type in {EVENT_TYPE_OPENED, EVENT_TYPE_CLOSED_NO_WRITE}:
# FIXME: see issue #949, and find a way to better handle that scenario
return
if self.event_debouncer is not None:
self.event_debouncer.handle_event(event)
else:
self._restart_process()
def _restart_process(self) -> None:
if self._is_trick_stopping:
return
self._stop_process()
self._start_process()
self.restart_count += 1
if platform.is_windows():
def kill_process(pid: int, stop_signal: int) -> None:
os.kill(pid, stop_signal)
else:
def kill_process(pid: int, stop_signal: int) -> None:
os.killpg(os.getpgid(pid), stop_signal)
|