forked from varia/varia.website
225 lines
7.5 KiB
Python
225 lines
7.5 KiB
Python
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)
|