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 threading
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))
                i = (e.inode(), e.stat().st_dev)

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

    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

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

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

    def mtime(self, path):
        """Return the last modified time of a file"""
        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

    def stat_snapshot(self):
        return {}

    def stat_info(self, path):
        return None

    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:

        t = threading.Timer(self.snapshot_secs, self.snapshot_timer)
        t.daemon = True

    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)


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

    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
            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)
            snapshot = self.load_snapshot(snapshot_file)

        return snapshot

    def load_snapshot(self, snapshot_file):
        """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:
            return pickle.load(f)

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

    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:

        for f in delta.files_deleted:

        for f in delta.files_modified:

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

    def add_path(self, path):
        """Adds a path for monitoring."""
        all = set([normalize(path)] + self.paths)
        unique = set(unique_path_roots(all))
        added = unique - set(self.paths)
        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:

        for p in added:

    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:

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

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

    def on_created(self, event):

    def on_modified(self, event):

    def on_deleted(self, 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)

        head, tail = os.path.split(path)
        while head and tail:
            if head in visited:
            head, tail = os.path.split(head)
            yield path

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

    def added(e):
        print('added: ', e)

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

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

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

        while True:
    except KeyboardInterrupt:
        for path in watcher.paths:
  • \$\begingroup\$ Welcome to codereview! This is a nice first question! \$\endgroup\$ Jan 26 '17 at 7:59
  • \$\begingroup\$ 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. \$\endgroup\$
    – esac
    Jan 26 '17 at 8:18
  • \$\begingroup\$ 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? \$\endgroup\$
    – Peilonrayz
    Jan 26 '17 at 8:58
  • \$\begingroup\$ 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. \$\endgroup\$
    – esac
    Jan 26 '17 at 16:59
  • \$\begingroup\$ I was just wondering if I had an excuse to try out the new pathlib library, ;P BTW os.scandir is 3.5+. \$\endgroup\$
    – Peilonrayz
    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.

  • \$\begingroup\$ 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..) \$\endgroup\$
    – esac
    Jan 29 '17 at 18:47
  • \$\begingroup\$ @esac: Any thoughts on #4? I'm curious as to your reasoning here. \$\endgroup\$ Jan 29 '17 at 18:49
  • \$\begingroup\$ 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. \$\endgroup\$
    – esac
    Jan 31 '17 at 22:39

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.