Create a watchdog timer in Python to look for changes in the filesystem



Many times a file must be processed during its creation or modification. This can be done by following the changes in a specific directory. There are many ways in python to keep track of changes made to a directory. One such way is to use the watchdog module. As the name suggests, this module monitors the given directory and can notify if a file is created or modified.

Required Modules

  • Guard Dog
    To install the watchdog, run this command in a terminal.
     pip install watchdog 
  • logging
    It comes with Python, so no need install it externally.

Now let`s dive into the code that will log all the logged changes. This code will only log changes checked in to the current directory.

# import modules

import sys

import time

import logging

from watchdog.observers import Observer

from watchdog.events import LoggingEventHandler

  

if __ name__ = = " __ main__ " :

  # Set the format for logging information

logging.basicConfig (level = logging.INFO,

format = `% (asctime) s -% (message) s` ,

datefmt = `% Y-% m-% d% H:% M:% S` )

 

# Install format for displaying the path

path = sys.argv [ 1 ] if len (sys.argv) & gt;  1 else `.`

 

# Initialize registration event handler

event_handler = LoggingEventHandler ()

 

# Initialize Observer

observer = Observer ()

observer.schedule (event_handler, path, recursive = True )

 

# Start observer

observer.start ()

try :

while True :

  # Set stream timeout

time.sleep ( 1 )

  except KeyboardInterrupt:

observer.stop ()

observer.join ()

Output:

The above code will log all changes registered to the directory. If you want to make changes while creating or modifying a file, you can do so using the following code.

# import time module, Observer, FileSystemEventHandler

import time

from watchdog.observers import Observer

from watchdog.events import FileSystemEventHandler

 

 

class OnMyWatch:

# Set directory to watch

watchDirectory = "/ give / the / address / of / directory"

 

def __ init __ ( self ):

self .observer = Observer ()

  

  def run ( self ):

event_handler = Handler ()

  self . observer.schedule (event_handler, self . watchDirectory, recursive = True )

  self . observer .start ()

try :

while True :

time.sleep ( 5 )

except :

  self . observer.stop ()

  print ( "Observer Stopped" )

  

  self . observer.join ()

 

 

class Handler (FileSystemEventHandler):

 

  @ staticmethod

def on_any_event (event):

if event.is_directory:

  return None

  

elif event.event_type = = `created` :

# The event was created, you can process it now

print ( "Watchdog received created event -% s." % event.src_path)

elif event.event_type = = `modified`

# The event has been changed, you can handle it now

print ( " Watchdog received modified event -% s. " % event.src_path)

 

 

if __ name__ = = ` __main__` :

watch = OnMyWatch ()

watch.run ()

Output: