Run process

run_process ¶

run_process(*paths: Union[Path, str], target: Union[str, Callable[..., Any]], args: Tuple[Any, ...] = (), kwargs: Optional[Dict[str, Any]] = None, target_type: Literal['function', 'command', 'auto'] = 'auto', callback: Optional[Callable[[Set[FileChange]], None]] = None, watch_filter: Optional[Callable[[Change, str], bool]] = DefaultFilter(), grace_period: float = 0, debounce: int = 1600, step: int = 50, debug: Optional[bool] = None, sigint_timeout: int = 5, sigkill_timeout: int = 1, recursive: bool = True, ignore_permission_denied: bool = False) -> int

Run a process and restart it upon file changes.

run_process can work in two ways:

Using multiprocessing.Process † to run a python function
Or, using subprocess.Popen to run a command

Note

† technically multiprocessing.get_context('spawn').Process to avoid forking and improve code reload/import.

Internally, run_process uses watch with raise_interrupt=False so the function exits cleanly upon Ctrl+C.

Parameters:

Name	Type	Description	Default
`*paths`	`Union[Path, str]`	matches the same argument of `watch`	`()`
`target`	`Union[str, Callable[..., Any]]`	function or command to run	required
`args`	`Tuple[Any, ...]`	arguments to pass to `target`, only used if `target` is a function	`()`
`kwargs`	`Optional[Dict[str, Any]]`	keyword arguments to pass to `target`, only used if `target` is a function	`None`
`target_type`	`Literal['function', 'command', 'auto']`	type of target. Can be `'function'`, `'command'`, or `'auto'` in which case `detect_target_type` is used to determine the type.	`'auto'`
`callback`	`Optional[Callable[[Set[FileChange]], None]]`	function to call on each reload, the function should accept a set of changes as the sole argument	`None`
`watch_filter`	`Optional[Callable[[Change, str], bool]]`	matches the same argument of `watch`	`DefaultFilter()`
`grace_period`	`float`	number of seconds after the process is started before watching for changes	`0`
`debounce`	`int`	matches the same argument of `watch`	`1600`
`step`	`int`	matches the same argument of `watch`	`50`
`debug`	`Optional[bool]`	matches the same argument of `watch`	`None`
`sigint_timeout`	`int`	the number of seconds to wait after sending sigint before sending sigkill	`5`
`sigkill_timeout`	`int`	the number of seconds to wait after sending sigkill before raising an exception	`1`
`recursive`	`bool`	matches the same argument of `watch`	`True`

Returns:

Type	Description
`int`	number of times the function was reloaded.

Example of run_process running a function

from watchfiles import run_process

def callback(changes):
    print('changes detected:', changes)

def foobar(a, b):
    print('foobar called with:', a, b)

if __name__ == '__main__':
    run_process('./path/to/dir', target=foobar, args=(1, 2), callback=callback)

As well as using a callback function, changes can be accessed from within the target function, using the WATCHFILES_CHANGES environment variable.

Example of run_process accessing changes

from watchfiles import run_process

def foobar(a, b, c):
    # changes will be an empty list "[]" the first time the function is called
    changes = os.getenv('WATCHFILES_CHANGES')
    changes = json.loads(changes)
    print('foobar called due to changes:', changes)

if __name__ == '__main__':
    run_process('./path/to/dir', target=foobar, args=(1, 2, 3))

Again with the target as command, WATCHFILES_CHANGES can be used to access changes.

example.sh

echo "changers: ${WATCHFILES_CHANGES}"

Example of run_process running a command

from watchfiles import run_process

if __name__ == '__main__':
    run_process('.', target='./example.sh')

arun_process `async` ¶

arun_process(*paths: Union[Path, str], target: Union[str, Callable[..., Any]], args: Tuple[Any, ...] = (), kwargs: Optional[Dict[str, Any]] = None, target_type: Literal['function', 'command', 'auto'] = 'auto', callback: Optional[Callable[[Set[FileChange]], Any]] = None, watch_filter: Optional[Callable[[Change, str], bool]] = DefaultFilter(), grace_period: float = 0, debounce: int = 1600, step: int = 50, debug: Optional[bool] = None, recursive: bool = True, ignore_permission_denied: bool = False) -> int

Async equivalent of run_process, all arguments match those of run_process except callback which can be a coroutine.

Starting and stopping the process and watching for changes is done in a separate thread.

As with run_process, internally arun_process uses awatch, however KeyboardInterrupt cannot be caught and suppressed in awatch so these errors need to be caught separately, see below.

Example of arun_process usage

import asyncio
from watchfiles import arun_process

async def callback(changes):
    await asyncio.sleep(0.1)
    print('changes detected:', changes)

def foobar(a, b):
    print('foobar called with:', a, b)

async def main():
    await arun_process('.', target=foobar, args=(1, 2), callback=callback)

if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print('stopped via KeyboardInterrupt')

detect_target_type ¶

detect_target_type(target: Union[str, Callable[..., Any]]) -> Literal['function', 'command']

Used by run_process, arun_process and indirectly the CLI to determine the target type with target_type is auto.

Detects the target type - either function or command. This method is only called with target_type='auto'.

The following logic is employed:

If target is not a string, it is assumed to be a function
If target ends with .py or .sh, it is assumed to be a command
Otherwise, the target is assumed to be a function if it matches the regex [a-zA-Z0-9_]+(\.[a-zA-Z0-9_]+)+

If this logic does not work for you, specify the target type explicitly using the target_type function argument or --target-type command line argument.