Detect file creation with watchdog

Question

I am trying to detect when a file with a given name is created in a directory. I am doing it thanks to watchdog. The creation is correctly detected but I don't know how to terminate the application properly once the detection is done.

My piece of code is the following:

#!/usr/bin/env python # -*- coding: utf-8 -*-  import logging import sys import time  from watchdog.events import FileSystemEventHandler from watchdog.observers import Observer  logging.basicConfig(level=logging.ERROR)  class MyEventHandler(FileSystemEventHandler):     def __init__(self, observer, filename):         self.observer = observer         self.filename = filename      def on_created(self, event):         print "e=", event         if not event.is_directory and event.src_path.endswith(self.filename):             print "file created"             self.observer.unschedule_all()             self.observer.stop()  def main(argv=None):     path = argv[1]     filename = argv[2]     observer = Observer()     event_handler = MyEventHandler(observer, filename)     observer.schedule(event_handler, path, recursive=False)     observer.start()     observer.join()     return 0  if __name__ == "__main__":     sys.exit(main(sys.argv)) 

I am new to python and I cannot figure out what is wrong. The detection seems to be scheduled in a dedicated thread and the join() method is waiting for this thread to terminate. Thus, I suppose that I am not calling the right method on the observer to stop waiting/looping, but the watchdog documentation seems really not clear to point out what are the methods that may be used.

Does someone have an idea how I can achieve my goal?

Answer 1

Finally, after taking a look at the watchdog implementation, it is not necessary to call unschedule_all before stop, this is done automatically. Removing the line containing this method call fixes the issue and the application is running perfectly.

Answer 2

Also, the below script is used to observe filename at a specific path using the PatternMatchingEventHandler.

import time from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler from watchdog.events import PatternMatchingEventHandler import sys   class Watcher:     def __init__(self, path, filename):         self.observer = Observer()         self.path = path         self.filename = filename      def run(self):         event_handler = Handler(self.filename)         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(PatternMatchingEventHandler):     def __init__(self, filename):         super(Handler, self).__init__(             patterns=[filename],             ignore_patterns=["*.tmp"],             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 = "."     filename = "test.csv"      w = Watcher(path, filename)     w.run() 

output:

[Tue Feb  9 01:55:38 2021] noticed: [created] on: [/Users/mt/Documents/stackoverflow/test.csv]  [Tue Feb  9 01:55:44 2021] noticed: [modified] on: [/Users/mt/Documents/stackoverflow/test.csv]  [Tue Feb  9 01:56:01 2021] noticed: [deleted] on: [/Users/mt/Documents/stackoverflow/test.csv]  

It is also possible to determine the creation of a new file without installing additional libraries.

import os import time   def watch_file(filename, time_limit=3600, check_interval=60):     """Return true if filename exists, if not keep checking once every check_interval seconds for time_limit seconds.     time_limit defaults to 1 hour     check_interval defaults to 1 minute     """     now = time.time()     last_time = now + time_limit          while time.time() <= last_time:         if os.path.exists(filename):             return True         else:             # Wait for check interval seconds, then check again.             time.sleep(check_interval)     return False   if __name__ == "__main__":     filename = "test.csv"     time_limit = 60     check_interval = 1      if watch_file(filename, time_limit, check_interval):         print(f"File created: {os.path.abspath(filename)}")     else:         print(             f"File {filename} not found after waiting: {time_limit} seconds!"         ) 

output:

File created: /Users/mt/Documents/stackoverflow/test.csv