codekingpro/portable-devtools
114k
1import contextlib
2import json
3import logging
4import os
5import re
6import shlex
7import signal
8import subprocess
9import sys
10from importlib import import_module
11from multiprocessing import get_context
12from multiprocessing.context import SpawnProcess
13from pathlib import Path
14from time import sleep
15from typing import TYPE_CHECKING, Any, Callable, Dict, Generator, List, Optional, Set, Tuple, Union
16
17import anyio
18
19from .filters import DefaultFilter
20from .main import Change, FileChange, awatch, watch
21
22if TYPE_CHECKING:
23 from typing import Literal
24
25__all__ = 'run_process', 'arun_process', 'detect_target_type', 'import_string'
26logger = logging.getLogger('watchfiles.main')
27
28
29def run_process(
30 *paths: Union[Path, str],
31 target: Union[str, Callable[..., Any]],
32 args: Tuple[Any, ...] = (),
33 kwargs: Optional[Dict[str, Any]] = None,
34 target_type: "Literal['function', 'command', 'auto']" = 'auto',
35 callback: Optional[Callable[[Set[FileChange]], None]] = None,
36 watch_filter: Optional[Callable[[Change, str], bool]] = DefaultFilter(),
37 grace_period: float = 0,
38 debounce: int = 1_600,
39 step: int = 50,
40 debug: Optional[bool] = None,
41 sigint_timeout: int = 5,
42 sigkill_timeout: int = 1,
43 recursive: bool = True,
44 ignore_permission_denied: bool = False,
45) -> int:
46 """
47 Run a process and restart it upon file changes.
48
49 `run_process` can work in two ways:
50
51 * Using `multiprocessing.Process` † to run a python function
52 * Or, using `subprocess.Popen` to run a command
53
54 !!! note
55
56 **†** technically `multiprocessing.get_context('spawn').Process` to avoid forking and improve
57 code reload/import.
58
59 Internally, `run_process` uses [`watch`][watchfiles.watch] with `raise_interrupt=False` so the function
60 exits cleanly upon `Ctrl+C`.
61
62 Args:
63 *paths: matches the same argument of [`watch`][watchfiles.watch]
64 target: function or command to run
65 args: arguments to pass to `target`, only used if `target` is a function
66 kwargs: keyword arguments to pass to `target`, only used if `target` is a function
67 target_type: type of target. Can be `'function'`, `'command'`, or `'auto'` in which case
68 [`detect_target_type`][watchfiles.run.detect_target_type] is used to determine the type.
69 callback: function to call on each reload, the function should accept a set of changes as the sole argument
70 watch_filter: matches the same argument of [`watch`][watchfiles.watch]
71 grace_period: number of seconds after the process is started before watching for changes
72 debounce: matches the same argument of [`watch`][watchfiles.watch]
73 step: matches the same argument of [`watch`][watchfiles.watch]
74 debug: matches the same argument of [`watch`][watchfiles.watch]
75 sigint_timeout: the number of seconds to wait after sending sigint before sending sigkill
76 sigkill_timeout: the number of seconds to wait after sending sigkill before raising an exception
77 recursive: matches the same argument of [`watch`][watchfiles.watch]
78
79 Returns:
80 number of times the function was reloaded.
81
82 ```py title="Example of run_process running a function"
83 from watchfiles import run_process
84
85 def callback(changes):
86 print('changes detected:', changes)
87
88 def foobar(a, b):
89 print('foobar called with:', a, b)
90
91 if __name__ == '__main__':
92 run_process('./path/to/dir', target=foobar, args=(1, 2), callback=callback)
93 ```
94
95 As well as using a `callback` function, changes can be accessed from within the target function,
96 using the `WATCHFILES_CHANGES` environment variable.
97
98 ```py title="Example of run_process accessing changes"
99 from watchfiles import run_process
100
101 def foobar(a, b, c):
102 # changes will be an empty list "[]" the first time the function is called
103 changes = os.getenv('WATCHFILES_CHANGES')
104 changes = json.loads(changes)
105 print('foobar called due to changes:', changes)
106
107 if __name__ == '__main__':
108 run_process('./path/to/dir', target=foobar, args=(1, 2, 3))
109 ```
110
111 Again with the target as `command`, `WATCHFILES_CHANGES` can be used
112 to access changes.
113
114 ```bash title="example.sh"
115 echo "changers: ${WATCHFILES_CHANGES}"
116 ```
117
118 ```py title="Example of run_process running a command"
119 from watchfiles import run_process
120
121 if __name__ == '__main__':
122 run_process('.', target='./example.sh')
123 ```
124 """
125 if target_type == 'auto':
126 target_type = detect_target_type(target)
127
128 logger.debug('running "%s" as %s', target, target_type)
129 catch_sigterm()
130 process = start_process(target, target_type, args, kwargs)
131 reloads = 0
132
133 if grace_period:
134 logger.debug('sleeping for %s seconds before watching for changes', grace_period)
135 sleep(grace_period)
136
137 try:
138 for changes in watch(
139 *paths,
140 watch_filter=watch_filter,
141 debounce=debounce,
142 step=step,
143 debug=debug,
144 raise_interrupt=False,
145 recursive=recursive,
146 ignore_permission_denied=ignore_permission_denied,
147 ):
148 callback and callback(changes)
149 process.stop(sigint_timeout=sigint_timeout, sigkill_timeout=sigkill_timeout)
150 process = start_process(target, target_type, args, kwargs, changes)
151 reloads += 1
152 finally:
153 process.stop()
154 return reloads
155
156
157async def arun_process(
158 *paths: Union[Path, str],
159 target: Union[str, Callable[..., Any]],
160 args: Tuple[Any, ...] = (),
161 kwargs: Optional[Dict[str, Any]] = None,
162 target_type: "Literal['function', 'command', 'auto']" = 'auto',
163 callback: Optional[Callable[[Set[FileChange]], Any]] = None,
164 watch_filter: Optional[Callable[[Change, str], bool]] = DefaultFilter(),
165 grace_period: float = 0,
166 debounce: int = 1_600,
167 step: int = 50,
168 debug: Optional[bool] = None,
169 recursive: bool = True,
170 ignore_permission_denied: bool = False,
171) -> int:
172 """
173 Async equivalent of [`run_process`][watchfiles.run_process], all arguments match those of `run_process` except
174 `callback` which can be a coroutine.
175
176 Starting and stopping the process and watching for changes is done in a separate thread.
177
178 As with `run_process`, internally `arun_process` uses [`awatch`][watchfiles.awatch], however `KeyboardInterrupt`
179 cannot be caught and suppressed in `awatch` so these errors need to be caught separately, see below.
180
181 ```py title="Example of arun_process usage"
182 import asyncio
183 from watchfiles import arun_process
184
185 async def callback(changes):
186 await asyncio.sleep(0.1)
187 print('changes detected:', changes)
188
189 def foobar(a, b):
190 print('foobar called with:', a, b)
191
192 async def main():
193 await arun_process('.', target=foobar, args=(1, 2), callback=callback)
194
195 if __name__ == '__main__':
196 try:
197 asyncio.run(main())
198 except KeyboardInterrupt:
199 print('stopped via KeyboardInterrupt')
200 ```
201 """
202 import inspect
203
204 if target_type == 'auto':
205 target_type = detect_target_type(target)
206
207 logger.debug('running "%s" as %s', target, target_type)
208 catch_sigterm()
209 process = await anyio.to_thread.run_sync(start_process, target, target_type, args, kwargs)
210 reloads = 0
211
212 if grace_period:
213 logger.debug('sleeping for %s seconds before watching for changes', grace_period)
214 await anyio.sleep(grace_period)
215
216 async for changes in awatch(
217 *paths,
218 watch_filter=watch_filter,
219 debounce=debounce,
220 step=step,
221 debug=debug,
222 recursive=recursive,
223 ignore_permission_denied=ignore_permission_denied,
224 ):
225 if callback is not None:
226 r = callback(changes)
227 if inspect.isawaitable(r):
228 await r
229
230 await anyio.to_thread.run_sync(process.stop)
231 process = await anyio.to_thread.run_sync(start_process, target, target_type, args, kwargs, changes)
232 reloads += 1
233 await anyio.to_thread.run_sync(process.stop)
234 return reloads
235
236
237# Use spawn context to make sure code run in subprocess
238# does not reuse imported modules in main process/context
239spawn_context = get_context('spawn')
240
241
242def split_cmd(cmd: str) -> List[str]:
243 import platform
244
245 posix = platform.uname().system.lower() != 'windows'
246 return shlex.split(cmd, posix=posix)
247
248
249def start_process(
250 target: Union[str, Callable[..., Any]],
251 target_type: "Literal['function', 'command']",
252 args: Tuple[Any, ...],
253 kwargs: Optional[Dict[str, Any]],
254 changes: Optional[Set[FileChange]] = None,
255) -> 'CombinedProcess':
256 if changes is None:
257 changes_env_var = '[]'
258 else:
259 changes_env_var = json.dumps([[c.raw_str(), p] for c, p in changes])
260
261 os.environ['WATCHFILES_CHANGES'] = changes_env_var
262
263 process: Union[SpawnProcess, subprocess.Popen[bytes]]
264 if target_type == 'function':
265 kwargs = kwargs or {}
266 if isinstance(target, str):
267 args = target, get_tty_path(), args, kwargs
268 target_ = run_function
269 kwargs = {}
270 else:
271 target_ = target
272
273 process = spawn_context.Process(target=target_, args=args, kwargs=kwargs)
274 process.start()
275 else:
276 if args or kwargs:
277 logger.warning('ignoring args and kwargs for "command" target')
278
279 assert isinstance(target, str), 'target must be a string to run as a command'
280 popen_args = split_cmd(target)
281 process = subprocess.Popen(popen_args)
282 return CombinedProcess(process)
283
284
285def detect_target_type(target: Union[str, Callable[..., Any]]) -> "Literal['function', 'command']":
286 """
287 Used by [`run_process`][watchfiles.run_process], [`arun_process`][watchfiles.arun_process]
288 and indirectly the CLI to determine the target type with `target_type` is `auto`.
289
290 Detects the target type - either `function` or `command`. This method is only called with `target_type='auto'`.
291
292 The following logic is employed:
293
294 * If `target` is not a string, it is assumed to be a function
295 * If `target` ends with `.py` or `.sh`, it is assumed to be a command
296 * Otherwise, the target is assumed to be a function if it matches the regex `[a-zA-Z0-9_]+(\\.[a-zA-Z0-9_]+)+`
297
298 If this logic does not work for you, specify the target type explicitly using the `target_type` function argument
299 or `--target-type` command line argument.
300
301 Args:
302 target: The target value
303
304 Returns:
305 either `'function'` or `'command'`
306 """
307 if not isinstance(target, str):
308 return 'function'
309 elif target.endswith(('.py', '.sh')):
310 return 'command'
311 elif re.fullmatch(r'[a-zA-Z0-9_]+(\.[a-zA-Z0-9_]+)+', target):
312 return 'function'
313 else:
314 return 'command'
315
316
317class CombinedProcess:
318 def __init__(self, p: 'Union[SpawnProcess, subprocess.Popen[bytes]]'):
319 self._p = p
320 assert self.pid is not None, 'process not yet spawned'
321
322 def stop(self, sigint_timeout: int = 5, sigkill_timeout: int = 1) -> None:
323 os.environ.pop('WATCHFILES_CHANGES', None)
324 if self.is_alive():
325 logger.debug('stopping process...')
326
327 os.kill(self.pid, signal.SIGINT)
328
329 try:
330 self.join(sigint_timeout)
331 except subprocess.TimeoutExpired:
332 # Capture this exception to allow the self.exitcode to be reached.
333 # This will allow the SIGKILL to be sent, otherwise it is swallowed up.
334 logger.warning('SIGINT timed out after %r seconds', sigint_timeout)
335 pass
336
337 if self.exitcode is None:
338 logger.warning('process has not terminated, sending SIGKILL')
339 os.kill(self.pid, signal.SIGKILL)
340 self.join(sigkill_timeout)
341 else:
342 logger.debug('process stopped')
343 else:
344 logger.warning('process already dead, exit code: %d', self.exitcode)
345
346 def is_alive(self) -> bool:
347 if isinstance(self._p, SpawnProcess):
348 return self._p.is_alive()
349 else:
350 return self._p.poll() is None
351
352 @property
353 def pid(self) -> int:
354 # we check the process has always been spawned when CombinedProcess is initialised
355 return self._p.pid # type: ignore[return-value]
356
357 def join(self, timeout: int) -> None:
358 if isinstance(self._p, SpawnProcess):
359 self._p.join(timeout)
360 else:
361 self._p.wait(timeout)
362
363 @property
364 def exitcode(self) -> Optional[int]:
365 if isinstance(self._p, SpawnProcess):
366 return self._p.exitcode
367 else:
368 return self._p.returncode
369
370
371def run_function(function: str, tty_path: Optional[str], args: Tuple[Any, ...], kwargs: Dict[str, Any]) -> None:
372 with set_tty(tty_path):
373 func = import_string(function)
374 func(*args, **kwargs)
375
376
377def import_string(dotted_path: str) -> Any:
378 """
379 Stolen approximately from django. Import a dotted module path and return the attribute/class designated by the
380 last name in the path. Raise ImportError if the import fails.
381 """
382 try:
383 module_path, class_name = dotted_path.strip(' ').rsplit('.', 1)
384 except ValueError as e:
385 raise ImportError(f'"{dotted_path}" doesn\'t look like a module path') from e
386
387 module = import_module(module_path)
388 try:
389 return getattr(module, class_name)
390 except AttributeError as e:
391 raise ImportError(f'Module "{module_path}" does not define a "{class_name}" attribute') from e
392
393
394def get_tty_path() -> Optional[str]: # pragma: no cover
395 """
396 Return the path to the current TTY, if any.
397
398 Virtually impossible to test in pytest, hence no cover.
399 """
400 try:
401 return os.ttyname(sys.stdin.fileno())
402 except OSError:
403 # fileno() always fails with pytest
404 return '/dev/tty'
405 except AttributeError:
406 # on Windows. No idea of a better solution
407 return None
408
409
410@contextlib.contextmanager
411def set_tty(tty_path: Optional[str]) -> Generator[None, None, None]:
412 if tty_path:
413 try:
414 with open(tty_path) as tty: # pragma: no cover
415 sys.stdin = tty
416 yield
417 except OSError:
418 # eg. "No such device or address: '/dev/tty'", see https://github.com/samuelcolvin/watchfiles/issues/40
419 yield
420 else:
421 # currently on windows tty_path is None and there's nothing we can do here
422 yield
423
424
425def raise_keyboard_interrupt(signum: int, _frame: Any) -> None: # pragma: no cover
426 logger.warning('received signal %s, raising KeyboardInterrupt', signal.Signals(signum))
427 raise KeyboardInterrupt
428
429
430def catch_sigterm() -> None:
431 """
432 Catch SIGTERM and raise KeyboardInterrupt instead. This means watchfiles will stop quickly
433 on `docker compose stop` and other cases where SIGTERM is sent.
434
435 Without this the watchfiles process will be killed while a running process will continue uninterrupted.
436 """
437 logger.debug('registering handler for SIGTERM on watchfiles process %d', os.getpid())
438 signal.signal(signal.SIGTERM, raise_keyboard_interrupt)
439