How to run an function when anything changes in a dir with Python Watchdog?

There are many ways in python to follow changes made in a directory. One such way is to use the watchdog module.

• Modules needed
  • To install watchdog run this command in the terminal.

pip install watchdog

If you want to make changes (currently in the root directory - ".") at the time a file/directory is created or modified, you can do so by using the following code:

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler


class Watcher:
    def __init__(self, path):
        self.observer = Observer()
        self.path = path

    def run(self):
        event_handler = Handler()
        self.observer.schedule(event_handler, self.path, recursive=True)
        self.observer.start()
        try:
            while True:
                time.sleep(1)
        except:
            self.observer.stop()
            print("Error")

        self.observer.join()


class Handler(FileSystemEventHandler):
    @staticmethod
    def on_any_event(event):
        # if event.is_directory:
        #     return None
        print(
            "[{}] noticed: [{}] on: [{}] ".format(
                time.asctime(), event.event_type, event.src_path
            )
        )


if __name__ == "__main__":
    w = Watcher(".")
    w.run()

• event.src_path will be the full file path
• event.event_type will be created, moved, etc.

If you want to ignore directory changes just remove the comment.

output:

[Tue Feb  9 00:16:02 2021] noticed: [created] on: [/Users/mt/Documents/stackoverflow/test.txt] 
[Tue Feb  9 00:16:02 2021] noticed: [modified] on: [/Users/mt/Documents/stackoverflow] 
[Tue Feb  9 00:16:19 2021] noticed: [created] on: [/Users/mt/Documents/stackoverflow/download.jpg] 
[Tue Feb  9 00:16:19 2021] noticed: [modified] on: [/Users/mt/Documents/stackoverflow] 
[Tue Feb  9 00:16:30 2021] noticed: [created] on: [/Users/mt/Documents/stackoverflow/new_folder] 
[Tue Feb  9 00:16:30 2021] noticed: [modified] on: [/Users/mt/Documents/stackoverflow] 
[Tue Feb  9 00:16:46 2021] noticed: [deleted] on: [/Users/mt/Documents/stackoverflow/new_folder] 
[Tue Feb  9 00:16:46 2021] noticed: [modified] on: [/Users/mt/Documents/stackoverflow] 
[Tue Feb  9 00:16:52 2021] noticed: [deleted] on: [/Users/mt/Documents/stackoverflow/download.jpg] 
[Tue Feb  9 00:16:52 2021] noticed: [modified] on: [/Users/mt/Documents/stackoverflow] 
[Tue Feb  9 00:17:00 2021] noticed: [deleted] on: [/Users/mt/Documents/stackoverflow/test.txt] 
[Tue Feb  9 00:17:00 2021] noticed: [modified] on: [/Users/mt/Documents/stackoverflow] 

The Observer is the class that watches for any file system change and then dispatches the event to the event handler. It monitors the file system and look for any changes.

The EventHandler is an object that will be notified when something happens to the file system. In general a script is written to watch over any type of new files created or modified like csv, txt, xml, jpg etc.

For example, in the code below the PatternMatchingEventHandler inherits from the FileSystemEventHandler class and is used to do just that. Some useful methods of this class are:

• on_any_event: will be executed for any event.
• on_created: Executed when a file or a directory is created.
• on_modified: Executed when a file is modified or a directory renamed.
• on_deleted: Executed when a file or directory is deleted.
• on_moved: Executed when a file or directory is moved.

The below script is used to observe only .csv files using the PatternMAtchingEventHandler. You can further extend the patterns list if you want to observe more than one type of file.

import watchdog.events
import watchdog.observers
import time
import sys


class Handler(watchdog.events.PatternMatchingEventHandler):
    def __init__(self):
        # Set the patterns for PatternMatchingEventHandler
        watchdog.events.PatternMatchingEventHandler.__init__(
            self,
            patterns=["*.csv"],
            ignore_directories=True,
            case_sensitive=False,
        )

    def on_any_event(self, event):
        print(
            "[{}] noticed: [{}] on: [{}] ".format(
                time.asctime(), event.event_type, event.src_path
            )
        )


if __name__ == "__main__":
    path = sys.argv[1] if len(sys.argv) > 1 else "."
    event_handler = Handler()
    observer = watchdog.observers.Observer()
    observer.schedule(event_handler, path=path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

• Here with the help of PatternMAtchingEventHandler we can take advantage of processing just events related with files with the .csv extension.

output:

[Tue Feb  9 00:18:51 2021] noticed: [created] on: [/Users/mt/Documents/stackoverflow/test.csv] 
[Tue Feb  9 00:18:59 2021] noticed: [modified] on: [/Users/mt/Documents/stackoverflow/test.csv] 
[Tue Feb  9 00:19:12 2021] noticed: [deleted] on: [/Users/mt/Documents/stackoverflow/test.csv] 

You need to subclass and do whatever you want in dispatch:

import sys
import time
import logging
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler

class Event(LoggingEventHandler):
    def dispatch(self, event):
        print("Foobar")

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 = Event()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

If you run the code you will see Foobar outputted whenever a change is detected, to ignore files you may need to use [events.PatternMatchingEventHandler][1]. There are various methods in each

To do something when a something is modified case we can override on_modified:

class Event(LoggingEventHandler):
    def on_modified(self, event):
        print("Doh")

And running the code using the class above with event_handler = Event() and changing a file will output something like:

Doh
Doh
Doh
Doh
Doh
Doh
Doh
2015-10-03 15:33:55 - Created file: ./test.txt___jb_bak___
2015-10-03 15:33:55 - Moved file: from ./test.txt to ./test.txt___jb_old___
2015-10-03 15:33:55 - Moved file: from ./test.txt___jb_bak___ to ./test.txt
2015-10-03 15:33:55 - Deleted file: ./test.txt___jb_old___
Doh

[1]: http://pythonhosted.org/watchdog/api.html#watchdog.events.PatternMatchingEventHandler EventHandler class you can override, it all depends on what it is you want to do. The LoggingEventHandler class itslef is a subclass of watchdog.events.FileSystemEventHandler:

class watchdog.events.FileSystemEventHandler Bases: object

Base file system event handler that you can override methods from.

dispatch(event) Dispatches events to the appropriate methods.

Parameters: event (FileSystemEvent) – The event object representing the file system event.

on_any_event(event) Catch-all event handler.

Parameters: event (FileSystemEvent) – The event object representing the file system event.

on_created(event) Called when a file or directory is created.

Parameters: event (DirCreatedEvent or FileCreatedEvent) – Event representing file/directory creation.

on_deleted(event) Called when a file or directory is deleted.

Parameters: event (DirDeletedEvent or FileDeletedEvent) – Event representing file/directory deletion.

on_modified(event) Called when a file or directory is modified.

Parameters: event (DirModifiedEvent or FileModifiedEvent) – Event representing file/directory modification.

on_moved(event) Called when a file or a directory is moved or renamed.

Parameters: event (DirMovedEvent or FileMovedEvent) – Event representing file/directory movement.