File system monitor which watches for file/directory changes even after the process is stopped

Building off the python FileSystemEventHandler and DirectorySnapshot classes, my goal is for a number of projects that I am working on to have a reusable library that will handle file system change notifications while my process is running, but also inform me of any changes that happened while my process was not running.

I implemented my own handler of DirectorySnapshot called DirectorySnapshotFast that utilizes os.scandir() for better performance .08 seconds with scandir() vs 15 seconds originally.

I also implemented a blank EmptyDirectorySnapshot (borrowed from a StackOverflow user and fixed up as it was not working), so that when no snapshot exists, everything is considered a new file to my program.

I also make sure to keep track of unique path roots - that means that I only monitor the topmost parent of a set of directories. So if it is being asked to monitor /home and then /home/user is added, it will not monitor for changes in /home/user because these will already be emitted by the first monitor.

This is my first real python code and I'll be glad to accept any criticism of style, naming, conventions, PEP, performance optimizations, etc.. I will post this on github in case there is any interest as a standalone project.

import time
import pickle
import locale
import os.path
from logger import logging
from event import Event
from os.path import realpath, normcase, sep, normpath
from functools import cmp_to_key
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, FileModifiedEvent, FileCreatedEvent, FileDeletedEvent, FileMovedEvent
from watchdog.utils.dirsnapshot import DirectorySnapshot, DirectorySnapshotDiff

logger = logging.getLogger('watcher')

class DirectorySnapshotFast(DirectorySnapshot):

def __init__(self, path):
"""Creates a directory snapshot by saving the file paths, inodes and device for every file recursively"""
self.recursive = True
self.listdir = None
self._stat_info = {}
self._inode_to_path = {}

st = os.stat(path)
self._stat_info[path] = st
self._inode_to_path[(st.st_ino, st.st_dev)] = path

for e in self.scandirs(path):
# use the drive letter on windows as st_dev is always 0 unless stat() is called
if os.name == 'nt':
i = (e.inode(), self._getdrive(e.path))
else:
i = (e.inode(), e.stat().st_dev)

self._inode_to_path[i] = e.path
self._stat_info[e.path] = e.stat()

@staticmethod
def _getdrive(path):
drive, tail = os.path.splitdrive(path)
return drive

def scandirs(self, top):
"""Recursively scan from the initial directory for all files and directories and return them as a generator"""
for e in os.scandir(top):
yield e

try:
is_dir = e.is_dir()
except OSError:
is_dir = False

if is_dir:
yield from self.scandirs(e.path)

def mtime(self, path):
return self._stat_info[path].st_mtime

class EmptyDirectorySnapshot(DirectorySnapshot):
"""
For use as the zeroth snapshot in a chain of DirectorySnapshotDiffs
"""

def path(self, id):
return None

def inode(self, path):
return None

def mtime(self, path):
return None

@property
def stat_snapshot(self):
return {}

def stat_info(self, path):
return None

@property
def paths(self):
return set()

# TODO: expire the snapshot after a certain time
# TODO: expire the snapshot if the machine name is different
class FileSystemWatcher(FileSystemEventHandler):
observers = {}

# we maintain a list of unique path roots for tracking that we are not trying to watch a sub-directory of a parent
paths = []
snapshot_filename = None
snapshot_secs = 300
created = Event()
modified = Event()
moved = Event()
deleted = Event()

def __init__(self, paths=[], snapshot_filename='.snapshot', snapshot_secs=300):
self.snapshot_secs = snapshot_secs
self.snapshot_filename = snapshot_filename
self.paths = list(unique_path_roots(paths))

def snapshot_timer(self, initial=False):
"""Start a timer that on the sub-sequent call will create snapshots for every monitored directory"""
if not initial:
for path in self.paths:
self.create_snapshot(path)

t.daemon = True
t.start()

def start(self):
"""Process snapshot deltas first, and then setup a file system monitor on them"""
for path in self.paths:
snapshot = self.snapshot(path)
self.write_snapshot(os.path.join(path, self.snapshot_filename), snapshot)
self.watch(path)

self.snapshot_timer(initial=True)

def stop(self):
"""Shutdown the watcher for all paths"""
for path, observer in self.observers.items():
self.paths.remove(path)
observer.stop()

def watch(self, path):
"""Starts the filesystem watcher on the specified path"""
if normalize(path) not in self.observers:
observer = Observer()
observer.schedule(self, path=path, recursive=True)
self.observers[normalize(path)] = observer
observer.start()
logger.debug('Watching directory: %s', path)

return self.observers[normalize(path)]

def snapshot(self, path):
"""Gets the previous snapshot, a new snapshot and compares them to process the deltas"""
snapshot = self.get_current_snapshot(path)
current = DirectorySnapshotFast(path)
self.process_deltas(snapshot, current)
return current

def create_snapshot(self, path):
"""Writes a snapshot to the specified path"""
logger.debug('Creating snapshot file for path: %s', path)
snapshot_file = os.path.join(path, self.snapshot_filename)
snapshot = DirectorySnapshotFast(path)
self.write_snapshot(snapshot_file, snapshot)

return snapshot

def write_snapshot(self, snapshot_file, snapshot):
"""Utility for writing the snapshot. Provide a handler/callback to write to sqlite or other in future"""
with open(snapshot_file, 'wb') as f:
pickle.dump(snapshot, f)

def get_current_snapshot(self, path):
"""Get what is considered the previous snapshot, this may be the Empty Snapshot which will only be empty once"""
snapshot_file = os.path.join(path, self.snapshot_filename)

# consider the snapshot empty if the file doesn't exist, but create it for the next call
if not os.path.isfile(snapshot_file):
snapshot = EmptyDirectorySnapshot(path)
self.create_snapshot(path)
else:

return snapshot

"""Utility to load the snapshot currently from disk, but can be set to sqlite or other in future"""
with open(snapshot_file, 'rb') as f:

@staticmethod
def snapshot_callback(self, stat_info):
"""Not used."""
pass

def process_deltas(self, previous, current):
"""Subtracts current snapshot from previous and calls event handlers for each type of event."""
delta = current - previous

for f in delta.files_created:
self.created(FileCreatedEvent(f))

for f in delta.files_deleted:
self.deleted(FileDeletedEvent(f))

for f in delta.files_modified:
self.modified(FileModifiedEvent(f))

for f in delta.files_moved:
(src, dest) = f
self.moved(FileMovedEvent(src, dest))

all = set([normalize(path)] + self.paths)
unique = set(unique_path_roots(all))
removed = set(self.paths) - unique

# handle removed unique roots first. these roots are now children of a parent that has been added
for p in removed:
self.remove_path(p)

self.paths.append(normalize(p))
self.watch(p)

def remove_path(self, path):
"""Remove the path from monitoring. Either it is no longer requested to be monitored, or it is now a child of a parent"""
if normalize(path) in self.paths:
self.paths.remove(normalize(path))

observer = self.observers.pop(normalize(path), None)
if observer:
observer.stop()

# overriden events for the observer. These are called by the super() class
def on_moved(self, event):
self.moved(event)

def on_created(self, event):
self.created(event)

def on_modified(self, event):
self.modified(event)

def on_deleted(self, event):
self.deleted(event)

def normalize(path):
"""Returns a lowercase (Windows) with extra chars removed path"""
return normcase(normpath(realpath(path)))

def unique_path_roots(paths):
"""Return a list of paths that exclude any children of the parent. For example, c:\\, c:\\users would only return c:\\"""
visited = set()
paths = list(set(paths))

for path in sorted(paths,key=cmp_to_key(locale.strcoll)):
path = normalize(path)

break
else:
yield path

if __name__ == '__main__':
def modded(e):
print('modified: ', e)

def deld(e):
print('deleted: ', e)

def movd(e):
print('moved: ', e)

watcher = FileSystemWatcher(paths=["c:\\users\\rayben\\pictures"])
watcher.modified += modded
watcher.deleted += deld
watcher.moved += movd
watcher.start()

try:
while True:
time.sleep(1)
except KeyboardInterrupt:
for path in watcher.paths:
watcher.create_snapshot(path)
watcher.stop()

• Welcome to codereview! This is a nice first question! Jan 26 '17 at 7:59
• Thanks! Definitely would like a look over before I release it, and being that I probably don't know the full feature set of python yet, there are probably better ways to do some of the things that I am doing. Looking for tips.
– esac
Jan 26 '17 at 8:18
• The way that you're using print indicates that you're primarily using Python 3. Are you targeting both Python 2 and Python 3, or a specific version of 3 and above? Jan 26 '17 at 8:58
• Only Python3 (and more specifically only ones that contain scandir()). If there is interest I could go back and change it, but it wouldn't be that difficult.
– esac
Jan 26 '17 at 16:59
• I was just wondering if I had an excuse to try out the new pathlib library, ;P BTW os.scandir is 3.5+. Jan 26 '17 at 21:28

I'm just going to review the DirectorySnapshotFast class.

1. There's no docstring for the class. What kind of thing is a DirectorySnapshotFast object? How is it different from a DirectorySnapshot object? Something to do with a faster implementation and different treatment of drives on Windows, but what exactly?

2. The __init__ method sets recursive and listdir attributes that are not used.

3. In the _inode_to_path dictionary you have different behaviour on Windows (using the drive letter) and non-Windows (using the device number) — except that for the original path you always use the device number. This seems to be a bug.

4. I have to confess that I don't understand the purpose of the different behaviour on Windows. A DirectorySnapshot object caches information about the files in a directory — and so all these files will have the same drive letter on Windows. But maybe there is something I missed? There needs to be an explanation somewhere of what problem is being solved here.

5. The test os.name == 'nt' is repeated for each directory e, but the result of this test is not going to change. Consider caching the result of this test in a local variable.

6. In the non-Windows case e.stat() is called twice for each entry. Consider caching this in a local variable.

7. The _getdrive method is called from only one place and is just one line long. I suggest inlining it at the point of use.

8. The scandirs method is only called from within __init__, so could be declared locally to that function (compare the implementation of DirectorySnapshot.__init__, where walk is declared locally.

9. OSError is a broad class of exceptions — consider catching just the exception that you want to ignore (presumably PermissionError).

10. The mtime method implementation is identical to the corresponding implemetation in the DirectorySnapshot class. There is no need to repeat this in the subclass.

• Thanks, I will be reworking the code as suggested. Right now I am stuck because there is a 'bug' in Py3.5 where the os.DirEntry object is not accessible. It was fixed in 3.6 but I am unable to move up at this point, so have to redo a couple of pieces of code. I don't necessarily agree with #5 but it will become a moo point with the rework anyway. As for #9 it has to be OSError because there are many reasons you may not have access to the file at the time (permission, locked, deleted/moved after opening it, etc..)
– esac
Jan 29 '17 at 18:47
• @esac: Any thoughts on #4? I'm curious as to your reasoning here. Jan 29 '17 at 18:49
• Sorry been travelling. I am basing it off the original implementation, so my only assumption is that if you follow symlinks, the file can actually reside on a different device/drive. I am going to test that out soon, just waiting for some time to sit down and rework all of your suggestions.
– esac
Jan 31 '17 at 22:39