From 3528ff9501c6c099199772cd5f4c2b98b1d6a78f Mon Sep 17 00:00:00 2001 From: Eli Uriegas Date: Sun, 19 Feb 2017 14:15:18 -0600 Subject: [PATCH] Add base event handler, utils function --- sanic/_live_reload.py | 48 +++++++++++++++++++++++++++++++++++++++++++ sanic/utils.py | 12 +++++++++++ 2 files changed, 60 insertions(+) create mode 100644 sanic/_live_reload.py diff --git a/sanic/_live_reload.py b/sanic/_live_reload.py new file mode 100644 index 00000000..38e2c9b1 --- /dev/null +++ b/sanic/_live_reload.py @@ -0,0 +1,48 @@ +import os +import signal +import time + +from watchdog.events import RegexMatchingEventHandler +from watchdog.observers import Observer + + +class SanicEventHandler(RegexMatchingEventHandler): + def __init__(self, interval=1): + super().__init__(regexes=['.*\.py', '.*\.pyc', '.*\.pyo']) + self.interval = interval + self.last_reload = self.current_time + + def on_created(self, event): + self.trigger_reload(event.src_path) + + def on_modified(self, event): + self.trigger_reload(event.src_path) + + def on_moved(self, event): + self.trigger_reload(event.src_path) + + def on_deleted(self, event): + self.trigger_reload(event.src_path) + + @property + def current_time(self): + return time.time() + + def trigger_reload(self, src_path): + if self.current_time - self.last_reload > self.interval: + os.kill(os.getpid(), signal.SIGHUP) + # TODO: Add a log for the src_path and reload mechanisms + self.last_reload = self.current_time + +if __name__ == '__main__': + import sys + handler = SanicEventHandler() + observer = Observer() + observer.schedule(event_handler=handler, path=sys.argv[1], recursive=True) + observer.start() + try: + while observer.isAlive(): + observer.join(1) + except KeyboardInterrupt: + observer.stop() + observer.join() diff --git a/sanic/utils.py b/sanic/utils.py index 6b97327b..3e79ca47 100644 --- a/sanic/utils.py +++ b/sanic/utils.py @@ -1,3 +1,5 @@ +import os +import inspect import warnings from sanic.testing import TestClient @@ -15,3 +17,13 @@ def sanic_endpoint_test(app, method='get', uri='/', gather_request=True, return test_client._sanic_endpoint_test( method, uri, gather_request, debug, server_kwargs, *request_args, **request_kwargs) + + +def get_original_caller_directory(): + """Gets the directory of the original caller for any function + + To be primarily used to identify where to watch for live reload situations + + :returns: The absolute path to the directory of the original caller + """ + return os.path.split(os.path.abspath(inspect.stack()[-1]))[0]