Skip to content

Instantly share code, notes, and snippets.

@devdave
Created June 4, 2019 15:21
Show Gist options
  • Save devdave/05de2ed2fa2aa0a09ba931db36314e3e to your computer and use it in GitHub Desktop.
Save devdave/05de2ed2fa2aa0a09ba931db36314e3e to your computer and use it in GitHub Desktop.
Slimmed down twisted compatible reloader
"""
A flask like reloader function for use with txweb
In user script it must follow the pattern
def main():
my main func that starts twisted
if __name__ == "__main__":
from txweb.sugar.reloader import reloader
reloader(main)
else:
TAC logic goes here but isn't necessary for short term dev work
Originally found via https://blog.elsdoerfer.name/2010/03/09/twisted-twistd-autoreload/
NOTE - pyutils appears to be dead or a completely different project
That link lead to this code snippet - https://bitbucket.org/miracle2k/pyutils/src/tip/pyutils/autoreload.py
I didn't like how the watching logic walked through sys.modules as I was just concerned with the immediate
project files and not the entire ecosystem. Instead it starts with the current working directory via os.getcwd
and then walks downward to look over .py files
sys.exit didn't work correctly so I switched to use os._exit as hardset. I am not sure what to do if
os._exit ever gets deprecated.
I also removed the checks for where to run the reloader logic and it will always run as a thread.
"""
import pathlib
import os
import sys
import time
try:
import thread
except ImportError:
try:
import _thread as thread
except ImportError:
try:
import dummy_thread as thread
except ImportError:
try:
import _dummy_thread as thread
except ImportError:
print("Alright... so I tried importing thread, that failed, so I tried _thread, that failed too")
print("..so then I tried dummy_thread, then _dummy_thread. All failed")
print(", at this point I am out of ideas here")
sys.exit(-1)
RUN_RELOADER = True
SENTINEL_CODE = 7211
SENTINEL_NAME = "RELOADER_ACTIVE"
SENTINEL_OS_EXIT = True
try:
"""
"Reason" is here https://code.djangoproject.com/ticket/2330
TODO - Figure out why threading needs to be imported as this feels like a problem
within stdlib.
"""
import threading
except ImportError:
pass
_watch_list = {}
_win = (sys.platform == "win32")
def build_list(root_dir, watch_self = False):
"""
Walk from root_dir down, collecting all files that end with ^*.py$ to watch
This could get into a recursive hell loop but I don't use symlinks in my projects
so just roll with it.
:param root_dir: pathlib.Path current working dir to search
:param watch_self: bool Watch the reloader script for changes, some insane dogfooding going on
:return: None
"""
global _watch_list
if watch_self is True:
selfpath = pathlib.Path(__file__)
stat = selfpath.stat()
_watch_list[selfpath] = (stat.st_size, stat.st_ctime, stat.st_mtime,)
for pathobj in root_dir.iterdir():
if pathobj.is_dir():
build_list(pathobj, watch_self=False)
elif pathobj.name.endswith(".py") and not (pathobj.name.endswith(".pyc") or pathobj.name.endswith(".pyo")):
stat = pathobj.stat()
_watch_list[pathobj] = (stat.st_size, stat.st_ctime, stat.st_mtime,)
else:
pass
def file_changed():
global _watch_list
change_detected = False
for pathname, (st_size, st_ctime, st_mtime) in _watch_list.items():
pathobj = pathlib.Path(pathname)
stat = pathobj.stat()
if pathobj.exists() is False:
raise Exception(f"Lost track of {pathname!r}")
elif stat.st_size != st_size:
change_detected = True
elif stat.st_ctime != st_ctime:
change_detected = True
elif _win is False and stat.st_mtime != st_mtime:
change_detected = True
if change_detected:
print(f"RELOADING - {pathobj} changed")
break
return change_detected
def watch_thread(os_exit = SENTINEL_OS_EXIT, watch_self=False):
exit_func = os._exit if os_exit is True else sys.exit
build_list(pathlib.Path(os.getcwd()), watch_self=watch_self)
while True:
if file_changed():
exit_func(SENTINEL_CODE)
time.sleep(1)
def run_reloader():
while True:
args = [sys.executable] + sys.argv
if _win:
args = ['"%s"' % arg for arg in args]
new_env = os.environ.copy()
new_env[SENTINEL_NAME] = "true"
print("Running reloader process")
exit_code = os.spawnve(os.P_WAIT, sys.executable, args, new_env)
if exit_code != SENTINEL_CODE:
return exit_code
def reloader_main(main_func, args, kwargs, watch_self=False):
"""
:param main_func:
:param args:
:param kwargs:
:return:
"""
# If it is, start watcher thread and then run the main_func in the parent process as thread 0
if os.environ.get(SENTINEL_NAME) == "true":
thread.start_new_thread(watch_thread, (), {"os_exit":SENTINEL_OS_EXIT,"watch_self":watch_self})
try:
main_func(*args, **kwargs)
except KeyboardInterrupt:
pass
else:
# respawn this script into a blocking subprocess
try:
sys.exit(run_reloader())
except KeyboardInterrupt:
#I should just raise this because its already broken free of its rails
pass
def reloader(main_func, args=None, kwargs=None, **more_options):
"""
To avoid fucking with twisted as much as possible, the watcher logic is shunted into
a thread while the main (twisted) reactor runs in the main thread.
:param main_func: The function to run in the main/primary thread
:param args: list of arguments
:param kwargs: dictionary of arguments
:param more_options: var trash currently
:return: None
"""
if args is None:
args = ()
if kwargs is None:
kwargs = {}
reloader_main(main_func, args, kwargs, **more_options)
"""
def main():
#startup twisted here
if __name__ == "__main__":
reloader(main)
"""
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment