Merge pull request #2 from samuelcolvin/cli

cli

README.rst

@@ -108,6 +108,32 @@ If these classes aren't sufficient you can define your own watcher, in particula
 you will want to override ``should_watch_dir`` and ``should_watch_file``. Unless you're
 doing something very odd you'll want to inherit from ``DefaultDirWatcher``.
 
+CLI
+...
+
+*wathgod* also comes with a CLI for running and reloading python code.
+
+Lets say you have ``foobar.py``:
+
+.. code:: python
+
+   from aiohttp import web
+
+   async def handle(request):
+       return web.Response(text='testing more')
+
+   app = web.Application()
+   app.router.add_get('/', handle)
+
+   def main():
+       web.run_app(app, port=8000)
+
+You could run this and reload it when any file in the current directory changes with::
+
+   watchgod foobar.main
+
+Run ``watchgod --help`` for more options. *watchgod* is also available as a python executable module
+via ``python -m watchgod ...``.
 
 Why no inotify / kqueue / fsevent / winapi support
 --------------------------------------------------
@@ -119,7 +145,7 @@ This is not an oversight, it's a decision with the following rationale:
 1. Polling is "fast enough", particularly since PEP 471 introduced fast ``scandir``.
 
    With a reasonably large project like the TutorCruncher code base with 850 files and 300k lines
-   of code *watchdog* can scan the entire tree in ~24ms. With a scan interval of 400ms that's roughly
+   of code *watchgod* can scan the entire tree in ~24ms. With a scan interval of 400ms that's roughly
    5% of one CPU - perfectly acceptable load during development.
 
 2. The clue is in the title, there are at least 4 different file notification systems to integrate

setup.cfg

@@ -21,6 +21,9 @@ precision = 2
 exclude_lines =
     pragma: no cover
     raise NotImplementedError
+omit =
+    # __main__.py is trivial and hard to test properly
+    */__main__.py
 
 [isort]
 line_length=120

setup.py

@@ -33,6 +33,10 @@
     author='Samuel Colvin',
     author_email='[email protected]',
     url='https://github.com/samuelcolvin/watchgod',
+    entry_points="""
+        [console_scripts]
+        watchgod=watchgod.cli:cli
+    """,
     license='MIT',
     packages=['watchgod'],
     python_requires='>=3.5',

tests/test_cli.py

@@ -0,0 +1,95 @@
+from pathlib import Path
+
+from watchgod.cli import callback, cli, run_function
+
+
+def foobar():
+    # used by tests below
+    Path('sentinel').write_text('ok')
+
+
+def test_simple(mocker, tmpdir):
+    mocker.patch('watchgod.cli.set_start_method')
+    mocker.patch('watchgod.cli.sys.stdin.fileno')
+    mocker.patch('os.ttyname', return_value='/path/to/tty')
+    mock_run_process = mocker.patch('watchgod.cli.run_process')
+    cli('tests.test_cli.foobar', str(tmpdir))
+    mock_run_process.assert_called_once_with(
+        Path(str(tmpdir)),
+        run_function,
+        args=('tests.test_cli.foobar', '/path/to/tty'),
+        callback=callback
+    )
+
+
+def test_invalid_import1(mocker, tmpdir, capsys):
+    sys_exit = mocker.patch('watchgod.cli.sys.exit')
+    cli('foobar')
+    sys_exit.assert_called_once_with(1)
+    out, err = capsys.readouterr()
+    assert out == ''
+    assert err == 'ImportError: "foobar" doesn\'t look like a module path\n'
+
+
+def test_invalid_import2(mocker, tmpdir, capsys):
+    sys_exit = mocker.patch('watchgod.cli.sys.exit')
+    cli('pprint.foobar')
+    sys_exit.assert_called_once_with(1)
+    out, err = capsys.readouterr()
+    assert out == ''
+    assert err == 'ImportError: Module "pprint" does not define a "foobar" attribute\n'
+
+
+def test_invalid_path(mocker, capsys):
+    sys_exit = mocker.patch('watchgod.cli.sys.exit')
+    cli('tests.test_cli.foobar', '/does/not/exist')
+    sys_exit.assert_called_once_with(1)
+    out, err = capsys.readouterr()
+    assert out == ''
+    assert err == 'path "/does/not/exist" is not a directory\n'
+
+
+def test_tty_os_error(mocker, tmpworkdir):
+    mocker.patch('watchgod.cli.set_start_method')
+    mocker.patch('watchgod.cli.sys.stdin.fileno', side_effect=OSError)
+    mock_run_process = mocker.patch('watchgod.cli.run_process')
+    cli('tests.test_cli.foobar')
+    mock_run_process.assert_called_once_with(
+        Path(str(tmpworkdir)),
+        run_function,
+        args=('tests.test_cli.foobar', '/dev/tty'),
+        callback=callback
+    )
+
+
+def test_tty_attribute_error(mocker, tmpdir):
+    mocker.patch('watchgod.cli.set_start_method')
+    mocker.patch('watchgod.cli.sys.stdin.fileno', side_effect=AttributeError)
+    mock_run_process = mocker.patch('watchgod.cli.run_process')
+    cli('tests.test_cli.foobar', str(tmpdir))
+    mock_run_process.assert_called_once_with(
+        Path(str(tmpdir)),
+        run_function,
+        args=('tests.test_cli.foobar', None),
+        callback=callback
+    )
+
+
+def test_run_function(tmpworkdir):
+    assert not tmpworkdir.join('sentinel').exists()
+    run_function('tests.test_cli.foobar', None)
+    assert tmpworkdir.join('sentinel').exists()
+
+
+def test_run_function_tty(tmpworkdir):
+    # could this cause problems by changing sys.stdin?
+    assert not tmpworkdir.join('sentinel').exists()
+    run_function('tests.test_cli.foobar', '/dev/tty')
+    assert tmpworkdir.join('sentinel').exists()
+
+
+def test_callback(mocker):
+    # boring we have to test this directly, but we do
+    mock_logger = mocker.patch('watchgod.cli.logger.info')
+    callback({1, 2, 3})
+    mock_logger.assert_called_once_with('%d files changed, reloading', 3)

watchgod/__main__.py

@@ -0,0 +1,4 @@
+from .cli import cli
+
+if __name__ == '__main__':
+    cli()

watchgod/cli.py

@@ -0,0 +1,95 @@
+import argparse
+import contextlib
+import logging
+import os
+import sys
+from importlib import import_module
+from multiprocessing import set_start_method
+from pathlib import Path
+from typing import Optional
+
+from watchgod import run_process
+
+logger = logging.getLogger('watchgod.cli')
+
+
+def import_string(dotted_path):
+    """
+    Stolen approximately from django. Import a dotted module path and return the attribute/class designated by the
+    last name in the path. Raise ImportError if the import fails.
+    """
+    try:
+        module_path, class_name = dotted_path.strip(' ').rsplit('.', 1)
+    except ValueError as e:
+        raise ImportError('"{}" doesn\'t look like a module path'.format(dotted_path)) from e
+
+    module = import_module(module_path)
+    try:
+        return getattr(module, class_name)
+    except AttributeError as e:
+        raise ImportError('Module "{}" does not define a "{}" attribute'.format(module_path, class_name)) from e
+
+
+@contextlib.contextmanager
+def set_tty(tty_path):
+    if tty_path:
+        with open(tty_path) as tty:
+            sys.stdin = tty
+            yield
+    else:
+        # currently on windows tty_path is None and there's nothing we can do here
+        yield
+
+
+def run_function(function: str, tty_path: Optional[str]):
+    with set_tty(tty_path):
+        func = import_string(function)
+        func()
+
+
+def callback(changes):
+    logger.info('%d files changed, reloading', len(changes))
+
+
+def cli(*args):
+    args = args or sys.argv[1:]
+    parser = argparse.ArgumentParser(
+        prog='watchgod',
+        description='Watch a directory and execute a python function on changes.'
+    )
+    parser.add_argument('function', help='Path to python function to execute.')
+    parser.add_argument('path', nargs='?', default='.', help='Filesystem path to watch, defaults to current directory.')
+    parser.add_argument('--verbosity', nargs='?', type=int, default=1, help='0, 1 (default) or 2')
+    arg_namespace = parser.parse_args(args)
+
+    log_level = {0: logging.WARNING, 1: logging.INFO, 2: logging.DEBUG}[arg_namespace.verbosity]
+    hdlr = logging.StreamHandler()
+    hdlr.setLevel(log_level)
+    hdlr.setFormatter(logging.Formatter(fmt='[%(asctime)s] %(message)s', datefmt='%H:%M:%S'))
+    wg_logger = logging.getLogger('watchgod')
+    wg_logger.addHandler(hdlr)
+    wg_logger.setLevel(log_level)
+
+    try:
+        import_string(arg_namespace.function)
+    except ImportError as e:
+        print('ImportError: {}'.format(e), file=sys.stderr)
+        return sys.exit(1)
+
+    path = Path(arg_namespace.path)
+    if not path.is_dir():
+        print('path "{}" is not a directory'.format(path), file=sys.stderr)
+        return sys.exit(1)
+    path = path.resolve()
+
+    try:
+        tty_path = os.ttyname(sys.stdin.fileno())
+    except OSError:
+        # fileno() always fails with pytest
+        tty_path = '/dev/tty'
+    except AttributeError:
+        # on windows. No idea of a better solution
+        tty_path = None
+    logger.info('watching "%s/" and reloading "%s" on changes...', path, arg_namespace.function)
+    set_start_method('spawn')
+    run_process(path, run_function, args=(arg_namespace.function, tty_path), callback=callback)

watchgod/main.py

@@ -103,7 +103,7 @@ def _start_process(target, args, kwargs):
 
 def _stop_process(process):
     if process.is_alive():
-        logger.debug('stopping server process...')
+        logger.debug('stopping process...')
         os.kill(process.pid, signal.SIGINT)
         process.join(5)
         if process.exitcode is None:
@@ -113,11 +113,11 @@ def _stop_process(process):
         else:
             logger.debug('process stopped')
     else:
-        logger.warning('server process already dead, exit code: %d', process.exitcode)
+        logger.warning('process already dead, exit code: %d', process.exitcode)
 
 
 def run_process(path: Union[Path, str], target: Callable, *,
-                args: Tuple[Any]=(),
+                args: Tuple=(),
                 kwargs: Dict[str, Any]=None,
                 callback: Callable[[Set[Tuple[Change, str]]], None]=None,
                 watcher_cls: Type[AllWatcher]=PythonWatcher,

watchgod/version.py

@@ -2,4 +2,4 @@
 
 __all__ = ['VERSION']
 
-VERSION = StrictVersion('0.0.2')
+VERSION = StrictVersion('0.0.3')