Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
import sys
import time
import logging
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler
class ObserverThread(Observer):
"""
Patched watched.Observer.
"""
def _stopThread(self, thread):
"""
Stop thread and wait for it to end.
"""
print "Stopping %s" % thread
thread.stop()
thread.join(60)
if thread.isAlive():
raise AssertionError('Failed to stop thread: %s' % thread)
def _clear_emitters(self):
for emitter in self._emitters:
self._stopThread(emitter)
self._emitters.clear()
self._emitter_for_watch.clear()
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
path = sys.argv[1] if len(sys.argv) > 1 else '.'
event_handler = LoggingEventHandler()
eevent_handler = LoggingEventHandler()
observer = ObserverThread()
ca = observer.schedule(event_handler, path, recursive=False)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
print "Stopping main thread..."
observer.stop()
observer.join()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.