Detect file creation with watchdog

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.

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

import timefrom watchdog.observers import Observerfrom watchdog.events import FileSystemEventHandlerfrom watchdog.events import PatternMatchingEventHandlerimport sysclass 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 osimport timedef 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 Falseif __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