cli.pyā¢7.71 kB
import argparse
import logging
import os
import shlex
import sys
from pathlib import Path
from textwrap import dedent
from typing import Any, Callable, List, Optional, Tuple, Union, cast
from . import Change
from .filters import BaseFilter, DefaultFilter, PythonFilter
from .run import detect_target_type, import_string, run_process
from .version import VERSION
logger = logging.getLogger('watchfiles.cli')
def resolve_path(path_str: str) -> Path:
path = Path(path_str)
if not path.exists():
raise FileNotFoundError(path)
else:
return path.resolve()
def cli(*args_: str) -> None:
"""
Watch one or more directories and execute either a shell command or a python function on file changes.
Example of watching the current directory and calling a python function:
watchfiles foobar.main
Example of watching python files in two local directories and calling a shell command:
watchfiles --filter python 'pytest --lf' src tests
See https://watchfiles.helpmanual.io/cli/ for more information.
"""
args = args_ or sys.argv[1:]
parser = argparse.ArgumentParser(
prog='watchfiles',
description=dedent((cli.__doc__ or '').strip('\n')),
formatter_class=argparse.RawTextHelpFormatter,
)
parser.add_argument('target', help='Command or dotted function path to run')
parser.add_argument(
'paths', nargs='*', default='.', help='Filesystem paths to watch, defaults to current directory'
)
parser.add_argument(
'--ignore-paths',
nargs='?',
type=str,
help=(
'Specify directories to ignore, '
'to ignore multiple paths use a comma as separator, e.g. "env" or "env,node_modules"'
),
)
parser.add_argument(
'--target-type',
nargs='?',
type=str,
default='auto',
choices=['command', 'function', 'auto'],
help=(
'Whether the target should be intercepted as a shell command or a python function, '
'defaults to "auto" which infers the target type from the target string'
),
)
parser.add_argument(
'--filter',
nargs='?',
type=str,
default='default',
help=(
'Which files to watch, defaults to "default" which uses the "DefaultFilter", '
'"python" uses the "PythonFilter", "all" uses no filter, '
'any other value is interpreted as a python function/class path which is imported'
),
)
parser.add_argument(
'--args',
nargs='?',
type=str,
help='Arguments to set on sys.argv before calling target function, used only if the target is a function',
)
parser.add_argument('--verbose', action='store_true', help='Set log level to "debug", wins over `--verbosity`')
parser.add_argument(
'--non-recursive', action='store_true', help='Do not watch for changes in sub-directories recursively'
)
parser.add_argument(
'--verbosity',
nargs='?',
type=str,
default='info',
choices=['warning', 'info', 'debug'],
help='Log level, defaults to "info"',
)
parser.add_argument(
'--sigint-timeout',
nargs='?',
type=int,
default=5,
help='How long to wait for the sigint timeout before sending sigkill.',
)
parser.add_argument(
'--grace-period',
nargs='?',
type=float,
default=0,
help='Number of seconds after the process is started before watching for changes.',
)
parser.add_argument(
'--sigkill-timeout',
nargs='?',
type=int,
default=1,
help='How long to wait for the sigkill timeout before issuing a timeout exception.',
)
parser.add_argument(
'--ignore-permission-denied',
action='store_true',
help='Ignore permission denied errors while watching files and directories.',
)
parser.add_argument('--version', '-V', action='version', version=f'%(prog)s v{VERSION}')
arg_namespace = parser.parse_args(args)
if arg_namespace.verbose:
log_level = logging.DEBUG
else:
log_level = getattr(logging, arg_namespace.verbosity.upper())
hdlr = logging.StreamHandler()
hdlr.setLevel(log_level)
hdlr.setFormatter(logging.Formatter(fmt='[%(asctime)s] %(message)s', datefmt='%H:%M:%S'))
wg_logger = logging.getLogger('watchfiles')
wg_logger.addHandler(hdlr)
wg_logger.setLevel(log_level)
if arg_namespace.target_type == 'auto':
target_type = detect_target_type(arg_namespace.target)
else:
target_type = arg_namespace.target_type
if target_type == 'function':
logger.debug('target_type=function, attempting import of "%s"', arg_namespace.target)
import_exit(arg_namespace.target)
if arg_namespace.args:
sys.argv = [arg_namespace.target] + shlex.split(arg_namespace.args)
elif arg_namespace.args:
logger.warning('--args is only used when the target is a function')
try:
paths = [resolve_path(p) for p in arg_namespace.paths]
except FileNotFoundError as e:
print(f'path "{e}" does not exist', file=sys.stderr)
sys.exit(1)
watch_filter, watch_filter_str = build_filter(arg_namespace.filter, arg_namespace.ignore_paths)
logger.info(
'watchfiles v%s š path=%s target="%s" (%s) filter=%s...',
VERSION,
', '.join(f'"{p}"' for p in paths),
arg_namespace.target,
target_type,
watch_filter_str,
)
run_process(
*paths,
target=arg_namespace.target,
target_type=target_type,
watch_filter=watch_filter,
debug=log_level == logging.DEBUG,
sigint_timeout=arg_namespace.sigint_timeout,
sigkill_timeout=arg_namespace.sigkill_timeout,
recursive=not arg_namespace.non_recursive,
ignore_permission_denied=arg_namespace.ignore_permission_denied,
grace_period=arg_namespace.grace_period,
)
def import_exit(function_path: str) -> Any:
cwd = os.getcwd()
if cwd not in sys.path:
sys.path.append(cwd)
try:
return import_string(function_path)
except ImportError as e:
print(f'ImportError: {e}', file=sys.stderr)
sys.exit(1)
def build_filter(
filter_name: str, ignore_paths_str: Optional[str]
) -> Tuple[Union[None, DefaultFilter, Callable[[Change, str], bool]], str]:
ignore_paths: List[Path] = []
if ignore_paths_str:
ignore_paths = [Path(p).resolve() for p in ignore_paths_str.split(',')]
if filter_name == 'default':
return DefaultFilter(ignore_paths=ignore_paths), 'DefaultFilter'
elif filter_name == 'python':
return PythonFilter(ignore_paths=ignore_paths), 'PythonFilter'
elif filter_name == 'all':
if ignore_paths:
logger.warning('"--ignore-paths" argument ignored as "all" filter was selected')
return None, '(no filter)'
watch_filter_cls = import_exit(filter_name)
if isinstance(watch_filter_cls, type) and issubclass(watch_filter_cls, DefaultFilter):
return watch_filter_cls(ignore_paths=ignore_paths), watch_filter_cls.__name__
if ignore_paths:
logger.warning('"--ignore-paths" argument ignored as filter is not a subclass of DefaultFilter')
if isinstance(watch_filter_cls, type) and issubclass(watch_filter_cls, BaseFilter):
return watch_filter_cls(), watch_filter_cls.__name__
else:
watch_filter = cast(Callable[[Change, str], bool], watch_filter_cls)
return watch_filter, repr(watch_filter_cls)