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?
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:
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']
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!