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?

Metropolis

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.

Metropolis

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

Ame