Skip to content

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(),
    debounce: int = 1600,
    step: int = 50,
    debug: bool = False,
    sigint_timeout: int = 5,
    sigkill_timeout: int = 1,
    recursive: bool = True
) -> 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()
debounce int

matches the same argument of watch

1600
step int

matches the same argument of watch

50
debug bool

matches the same argument of watch

False
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(),
    debounce: int = 1600,
    step: int = 50,
    debug: bool = False,
    recursive: bool = True
) -> 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.

Parameters:

Name Type Description Default
target Union[str, Callable[..., Any]]

The target value

required

Returns:

Type Description
Literal['function', 'command']

either 'function' or 'command'