codekingpro/portable-devtools

1import argparse
2import logging
3import os
4import shlex
5import sys
6from pathlib import Path
7from textwrap import dedent
8from typing import Any, Callable, List, Optional, Tuple, Union, cast
9
10from . import Change
11from .filters import BaseFilter, DefaultFilter, PythonFilter
12from .run import detect_target_type, import_string, run_process
13from .version import VERSION
14
15logger = logging.getLogger('watchfiles.cli')
16
17
18def resolve_path(path_str: str) -> Path:
19    path = Path(path_str)
20    if not path.exists():
21        raise FileNotFoundError(path)
22    else:
23        return path.resolve()
24
25
26def cli(*args_: str) -> None:
27    """
28    Watch one or more directories and execute either a shell command or a python function on file changes.
29
30    Example of watching the current directory and calling a python function:
31
32        watchfiles foobar.main
33
34    Example of watching python files in two local directories and calling a shell command:
35
36        watchfiles --filter python 'pytest --lf' src tests
37
38    See https://watchfiles.helpmanual.io/cli/ for more information.
39    """
40    args = args_ or sys.argv[1:]
41    parser = argparse.ArgumentParser(
42        prog='watchfiles',
43        description=dedent((cli.__doc__ or '').strip('\n')),
44        formatter_class=argparse.RawTextHelpFormatter,
45    )
46    parser.add_argument('target', help='Command or dotted function path to run')
47    parser.add_argument(
48        'paths', nargs='*', default='.', help='Filesystem paths to watch, defaults to current directory'
49    )
50
51    parser.add_argument(
52        '--ignore-paths',
53        nargs='?',
54        type=str,
55        help=(
56            'Specify directories to ignore, '
57            'to ignore multiple paths use a comma as separator, e.g. "env" or "env,node_modules"'
58        ),
59    )
60    parser.add_argument(
61        '--target-type',
62        nargs='?',
63        type=str,
64        default='auto',
65        choices=['command', 'function', 'auto'],
66        help=(
67            'Whether the target should be intercepted as a shell command or a python function, '
68            'defaults to "auto" which infers the target type from the target string'
69        ),
70    )
71    parser.add_argument(
72        '--filter',
73        nargs='?',
74        type=str,
75        default='default',
76        help=(
77            'Which files to watch, defaults to "default" which uses the "DefaultFilter", '
78            '"python" uses the "PythonFilter", "all" uses no filter, '
79            'any other value is interpreted as a python function/class path which is imported'
80        ),
81    )
82    parser.add_argument(
83        '--args',
84        nargs='?',
85        type=str,
86        help='Arguments to set on sys.argv before calling target function, used only if the target is a function',
87    )
88    parser.add_argument('--verbose', action='store_true', help='Set log level to "debug", wins over `--verbosity`')
89    parser.add_argument(
90        '--non-recursive', action='store_true', help='Do not watch for changes in sub-directories recursively'
91    )
92    parser.add_argument(
93        '--verbosity',
94        nargs='?',
95        type=str,
96        default='info',
97        choices=['warning', 'info', 'debug'],
98        help='Log level, defaults to "info"',
99    )
100    parser.add_argument(
101        '--sigint-timeout',
102        nargs='?',
103        type=int,
104        default=5,
105        help='How long to wait for the sigint timeout before sending sigkill.',
106    )
107    parser.add_argument(
108        '--grace-period',
109        nargs='?',
110        type=float,
111        default=0,
112        help='Number of seconds after the process is started before watching for changes.',
113    )
114    parser.add_argument(
115        '--sigkill-timeout',
116        nargs='?',
117        type=int,
118        default=1,
119        help='How long to wait for the sigkill timeout before issuing a timeout exception.',
120    )
121    parser.add_argument(
122        '--ignore-permission-denied',
123        action='store_true',
124        help='Ignore permission denied errors while watching files and directories.',
125    )
126    parser.add_argument('--version', '-V', action='version', version=f'%(prog)s v{VERSION}')
127    arg_namespace = parser.parse_args(args)
128
129    if arg_namespace.verbose:
130        log_level = logging.DEBUG
131    else:
132        log_level = getattr(logging, arg_namespace.verbosity.upper())
133
134    hdlr = logging.StreamHandler()
135    hdlr.setLevel(log_level)
136    hdlr.setFormatter(logging.Formatter(fmt='[%(asctime)s] %(message)s', datefmt='%H:%M:%S'))
137    wg_logger = logging.getLogger('watchfiles')
138    wg_logger.addHandler(hdlr)
139    wg_logger.setLevel(log_level)
140
141    if arg_namespace.target_type == 'auto':
142        target_type = detect_target_type(arg_namespace.target)
143    else:
144        target_type = arg_namespace.target_type
145
146    if target_type == 'function':
147        logger.debug('target_type=function, attempting import of "%s"', arg_namespace.target)
148        import_exit(arg_namespace.target)
149        if arg_namespace.args:
150            sys.argv = [arg_namespace.target] + shlex.split(arg_namespace.args)
151    elif arg_namespace.args:
152        logger.warning('--args is only used when the target is a function')
153
154    try:
155        paths = [resolve_path(p) for p in arg_namespace.paths]
156    except FileNotFoundError as e:
157        print(f'path "{e}" does not exist', file=sys.stderr)
158        sys.exit(1)
159
160    watch_filter, watch_filter_str = build_filter(arg_namespace.filter, arg_namespace.ignore_paths)
161
162    logger.info(
163        'watchfiles v%s 👀  path=%s target="%s" (%s) filter=%s...',
164        VERSION,
165        ', '.join(f'"{p}"' for p in paths),
166        arg_namespace.target,
167        target_type,
168        watch_filter_str,
169    )
170
171    run_process(
172        *paths,
173        target=arg_namespace.target,
174        target_type=target_type,
175        watch_filter=watch_filter,
176        debug=log_level == logging.DEBUG,
177        sigint_timeout=arg_namespace.sigint_timeout,
178        sigkill_timeout=arg_namespace.sigkill_timeout,
179        recursive=not arg_namespace.non_recursive,
180        ignore_permission_denied=arg_namespace.ignore_permission_denied,
181        grace_period=arg_namespace.grace_period,
182    )
183
184
185def import_exit(function_path: str) -> Any:
186    cwd = os.getcwd()
187    if cwd not in sys.path:
188        sys.path.append(cwd)
189
190    try:
191        return import_string(function_path)
192    except ImportError as e:
193        print(f'ImportError: {e}', file=sys.stderr)
194        sys.exit(1)
195
196
197def build_filter(
198    filter_name: str, ignore_paths_str: Optional[str]
199) -> Tuple[Union[None, DefaultFilter, Callable[[Change, str], bool]], str]:
200    ignore_paths: List[Path] = []
201    if ignore_paths_str:
202        ignore_paths = [Path(p).resolve() for p in ignore_paths_str.split(',')]
203
204    if filter_name == 'default':
205        return DefaultFilter(ignore_paths=ignore_paths), 'DefaultFilter'
206    elif filter_name == 'python':
207        return PythonFilter(ignore_paths=ignore_paths), 'PythonFilter'
208    elif filter_name == 'all':
209        if ignore_paths:
210            logger.warning('"--ignore-paths" argument ignored as "all" filter was selected')
211        return None, '(no filter)'
212
213    watch_filter_cls = import_exit(filter_name)
214    if isinstance(watch_filter_cls, type) and issubclass(watch_filter_cls, DefaultFilter):
215        return watch_filter_cls(ignore_paths=ignore_paths), watch_filter_cls.__name__
216
217    if ignore_paths:
218        logger.warning('"--ignore-paths" argument ignored as filter is not a subclass of DefaultFilter')
219
220    if isinstance(watch_filter_cls, type) and issubclass(watch_filter_cls, BaseFilter):
221        return watch_filter_cls(), watch_filter_cls.__name__
222    else:
223        watch_filter = cast(Callable[[Change, str], bool], watch_filter_cls)
224        return watch_filter, repr(watch_filter_cls)
225