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

bryccw commented Jun 24, 2019

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

Copy link

SumedhaZestl commented Sep 29, 2019

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