Using watchdog to put newly created file names into variables

Question

I would like to use watchdog to find new files that are created in one folder. The file name will then be used in a different function. I am using this code here:

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

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO,
                        format=' %(message)s')
    path = sys.argv[1] if len(sys.argv) > 1 else '.'
    event_handler = LoggingEventHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

This gives the output for example in the console:

Created file: ./test_file.h5
Modified directory: .
Modified file: ./test_file.h5
Modified directory: .

What I would like is only when new files are created, that the name of them be returned and I don't need it returned in the console but just rather returned in a variable so I can use it as an input for a different function. Is there a way to do this?

Answer 1

You need to create custom handler, it can be done by inheriting FileSystemEventHandler and overriding event you want to use.

class CustomHandler(FileSystemEventHandler):

    def __init__(self, callback: Callable):
        self.callback = callback

    def on_created(self, event: Union[DirCreatedEvent, FileCreatedEvent]):
        print(f"Event type: {event.event_type}\nAt: {event.src_path}\n")

        if isinstance(event, FileCreatedEvent):
            file = pathlib.Path(event.src_path)

            print(f"Processing file {file.name}\n")

            self.callback(file)

Available events are:

• on_modified(self, event)
• on_deleted(self, event)
• on_closed(self, event)
• on_moved(self, event)
• on_any_event(self, event)

Each event could vary, for on_created - it would be safe to assume it's only DirCreatedEvent and FileCreatedEvent.

--

Example code

import time
import pathlib
from typing import Union

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, DirCreatedEvent, FileCreatedEvent


class CustomHandler(FileSystemEventHandler):
    """Custom handler for Watchdog"""

    def __init__(self):
        # List to store path
        self.path_strings = []

    # callback for File/Directory created event, called by Observer.
    def on_created(self, event: Union[DirCreatedEvent, FileCreatedEvent]):
        print(f"Event type: {event.event_type}\nAt: {event.src_path}")

        # check if it's File creation, not Directory creation
        if isinstance(event, FileCreatedEvent):
            
            # if so, do something with event.src_path - it's path of the created file.
            self.path_strings.append(event.src_path)

            print(f"Path content: \n{self.path_strings}")


def main():
    # get current path as absolute, linux-style path.
    working_path = pathlib.Path(".").absolute().as_posix()

    # create instance of observer and CustomHandler
    observer = Observer()
    handler = CustomHandler()

    # start observer, checks files recursively
    observer.schedule(handler, path=working_path, recursive=True)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()


if __name__ == '__main__':
    main()

Example output

Event type: created
At: E:/github/ProjectIncubator/single_run_scripts\test.a
Path content: 
['E:/github/ProjectIncubator/single_run_scripts\\test.a']
Event type: created
At: E:/github/ProjectIncubator/single_run_scripts\nyan.txt
Path content: 
['E:/github/ProjectIncubator/single_run_scripts\\test.a', 'E:/github/ProjectIncubator/single_run_scripts\\nyan.txt']

---

Old

Full Example Code

import time
import pathlib
import argparse
from typing import Union, Callable

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, DirCreatedEvent, FileCreatedEvent


class CustomHandler(FileSystemEventHandler):

    def __init__(self, callback: Callable):
        self.callback = callback

        # Store callback to be called on every on_created event

    def on_created(self, event: Union[DirCreatedEvent, FileCreatedEvent]):
        print(f"Event type: {event.event_type}\nAt: {event.src_path}\n")

        # check if it's File creation, not Directory creation
        if isinstance(event, FileCreatedEvent):
            file = pathlib.Path(event.src_path)

            print(f"Processing file {file.name}\n")

            # call callback
            self.callback(file)


def main():
    path: pathlib.Path = args.dir

    # list for new files
    created_files = []

    # create callback
    def callback(path_: pathlib.Path):
        print(f"Adding {path_.name} to list!")
        created_files.append(path_)

    # create instance of observer and CustomHandler
    observer = Observer()
    handler = CustomHandler(callback)

    observer.schedule(handler, path=path.absolute().as_posix(), recursive=True)
    observer.start()

    print("Observer started")

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

    print(f"{len(created_files)} new files was created!", "\n".join(p.name for p in created_files), sep="\n")
    input("Press enter to exit!")


if __name__ == '__main__':

    # get script's root
    ROOT = pathlib.Path(__file__).parent

    # parse argument - if provided given directory is used, otherwise 
    parser = argparse.ArgumentParser(description="Listen for file change in directory.")

    parser.add_argument("dir", metavar="DIR", type=pathlib.Path, default=ROOT, nargs="?", help="Directory to listen for. If omitted, script path is used.")

    args = parser.parse_args()

    main()

---

Example output (Line breaks are mess, sorry!)

❯ py .\watchdog_test.py X:\test
Observer started
Event type: created
At: X:/test\새 폴더

Event type: created
At: X:/test\meow.txt

Processing file meow.txt

Adding meow.txt to list!
Event type: created
At: X:/test\meow.txt

Processing file meow.txt

Adding meow.txt to list!
Event type: created
At: X:/test\meow - 복사본.txt

Processing file meow - 복사본.txt

Adding meow - 복사본.txt to list!
Event type: created
At: X:/test\meow - 복사본 (2).txt

Processing file meow - 복사본 (2).txt

Adding meow - 복사본 (2).txt to list!
Event type: created
At: X:/test\meow - 복사본 (3).txt

Processing file meow - 복사본 (3).txt

Adding meow - 복사본 (3).txt to list!
5 new files was created!
meow.txt
meow.txt
meow - 복사본.txt
meow - 복사본 (2).txt
meow - 복사본 (3).txt
Press enter to exit!