Skip to content

Instantly share code, notes, and snippets.

@justquick
Created August 3, 2012 16:33
Show Gist options
  • Save justquick/3249273 to your computer and use it in GitHub Desktop.
Save justquick/3249273 to your computer and use it in GitHub Desktop.
twitter logging example using werkzeug's reloader
import os
import sys
import signal
import json
import requests
from subprocess import call
from time import time, sleep
from itertools import chain
try:
from urllib import urlopen
except ImportError:
from urllib.request import urlopen
try:
from thread import start_new_thread
except ImportError:
from _thread import start_new_thread
def reloader_loop(extra_files=None, interval=1):
"""When this function is run from the main thread, it will force other
threads to exit when any modules currently loaded change.
Copyright notice. This function is based on the autoreload.py from
the CherryPy trac which originated from WSGIKit which is now dead.
:param extra_files: a list of additional files it should watch.
"""
def _iter_module_files():
modules = list(sys.modules.values())
for module in modules:
filename = getattr(module, '__file__', None)
if filename:
old = None
while not os.path.isfile(filename):
old = filename
filename = os.path.dirname(filename)
if filename == old:
break
else:
if filename[-4:] in ('.pyc', '.pyo'):
filename = filename[:-1]
yield filename
mtimes = {}
while 1:
for filename in chain(_iter_module_files(), extra_files or ()):
try:
mtime = os.stat(filename).st_mtime
except OSError:
continue
old_time = mtimes.get(filename)
if old_time is None:
mtimes[filename] = mtime
continue
elif mtime > old_time:
sys.exit(3)
sleep(interval)
def restart_with_reloader():
"""Spawn a new Python interpreter with the same arguments as this one,
but running the reloader thread.
"""
while 1:
print(' * Restarting with reloader')
environ = os.environ.copy()
environ['RUN_MAIN'] = 'true'
exit_code = call([sys.executable] + sys.argv, env=environ)
if exit_code != 3:
return exit_code
def run_with_reloader(main_func, extra_files=None, interval=1, args=(), kwargs={}):
"""Run the given function in an independent python interpreter."""
if os.environ.get('RUN_MAIN', '') == 'true':
start_new_thread(main_func, args, kwargs)
try:
reloader_loop(extra_files, interval)
except KeyboardInterrupt:
return
else:
try:
status = restart_with_reloader()
signal.signal(signal.SIGTERM, lambda *args: sys.exit(0))
sys.exit(status)
except KeyboardInterrupt:
pass
def main(auth=('foo','bar')):
resp = requests.post('https://stream.twitter.com/1/statuses/sample.json',
auth=auth)
with open('%s.log'%auth[0], 'a') as f:
f.write('running\n')
for tweet in resp.iter_lines():
if tweet:
f.write('%s\n' % json.loads(tweet.decode('utf-8')).get('id', ''))
#except ValueError: raise
if __name__ == '__main__':
start_new_thread(main)
run_with_reloader(main)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment