Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
Persistent watchdog observer. When watchdog re-start, check if new/modify/delete/etc.. files or directories since the last launch, and send events for suscribers handlers.
#!/usr/bin/env python
# -*- coding: utf-8 -*-
Subclassing Observer for saving states of folders, and load this states at the next observation.
TODO : mapping events and handlers dispatching, for a shorter code.
from __future__ import unicode_literals, print_function, division
import cPickle as pickle
import os
from watchdog.observers import Observer
from watchdog.utils.dirsnapshot import DirectorySnapshot, DirectorySnapshotDiff
from import FileCreatedEvent, FileDeletedEvent, FileModifiedEvent, FileMovedEvent
from import DirCreatedEvent, DirDeletedEvent, DirModifiedEvent, DirMovedEvent
__author__ = "Serge Kilimoff-Goriatchkine"
__licence__ = 'MIT Licence'
class _EmptySnapshot(object):
def stat_snapshot(self):
return dict()
def paths(self):
return set()
class PersistantObserver(Observer):
def __init__(self, *args, **kwargs):
Check if watching folders has changed since last observation.
If change detected, emit corresponding events at suscribers handlers.
At the `Observer.stop`, save states of folders with pickle for the next observation.
save_to : unicode
path where save pickle dumping
protocol (optionnal): int
protocol used for dump current states of watching folders
self._filename = kwargs.pop('save_to')
self._protocol = kwargs.pop('protocol', 0)
Observer.__init__(self, *args, **kwargs)
def start(self, *args, **kwargs):
previous_snapshots = dict()
if os.path.exists(self._filename):
with open(self._filename) as f:
previous_snapshots = pickle.load(f)
for watcher, handlers in self._handlers.iteritems():
path = watcher.path
curr_snap = DirectorySnapshot(path)
pre_snap = previous_snapshots.get(path, _EmptySnapshot())
diff = DirectorySnapshotDiff(pre_snap, curr_snap)
for handler in handlers:
# Dispatch files modifications
for new_path in diff.files_created:
for del_path in diff.files_deleted:
for mod_path in diff.files_modified:
for mov_path in diff.files_moved:
# Dispatch directories modifications
for new_dir in diff.dirs_created:
for del_dir in diff.dirs_deleted:
for mod_dir in diff.dirs_modified:
for mov_dir in diff.dirs_moved:
Observer.start(self, *args, **kwargs)
def stop(self, *args, **kwargs):
snapshots = {handler.path : DirectorySnapshot(handler.path) for handler in self._handlers.iterkeys()}
with open(self._filename, 'wb') as f:
pickle.dump(snapshots, f, self._protocol)
Observer.stop(self, *args, **kwargs)
if __name__ == "__main__":
# Simple exemple, derivated from watchdog doc.
import logging
from import LoggingEventHandler
event_handler = LoggingEventHandler()
observer = PersistantObserver(save_to='/tmp/test.pickle', protocol=-1)
observer.schedule(event_handler, path='/tmp/test', recursive=True)
while True:
except KeyboardInterrupt:
Copy link

uncw commented Jun 24, 2019

You will probably need to use dill package to pickle lambda functions.

Copy link

Getting following error on my first execution of a code

Traceback (most recent call last):
File "", line 117, in
File "", line 80, in start
diff = DirectorySnapshotDiff(pre_snap, curr_snap)
File "/usr/local/lib/python2.7/site-packages/watchdog/utils/", line 91, in init
old_path = ref.path(inode)
AttributeError: '_EmptySnapshot' object has no attribute 'path'

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment