Keep track of the file with asyncio

Question

I am trying to find a good way to observe the appearance of a file using the Python asyncio library . This is what I came up with:

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Watches for the appearance of a file.""" import argparse import asyncio import os.path @asyncio.coroutine def watch_for_file(file_path, interval=1): while True: if not os.path.exists(file_path): print("{} not found yet.".format(file_path)) yield from asyncio.sleep(interval) else: print("{} found!".format(file_path)) break def make_cli_parser(): cli_parser = argparse.ArgumentParser(description=__doc__) cli_parser.add_argument('file_path') return cli_parser def main(argv=None): cli_parser = make_cli_parser() args = cli_parser.parse_args(argv) loop = asyncio.get_event_loop() loop.run_until_complete(watch_for_file(args.file_path)) if __name__ == '__main__': main() 

I saved this as watch_for_file.py and can run it with

 python3 watch_for_file.py testfile 

In another shell session, I issue

 touch testfile 

to end the cycle.

Is there a more elegant solution than using this infinite loop and yield from asyncio.sleep() ?

Answer 1

Well, there are nicer platform-specific notification methods when creating a file. Gerrat is linked to one for Windows in his comment, and pyinotify can be used for Linux. Perhaps these platform approaches can be connected to asyncio , but you would end up writing a whole bunch of code to make it work regardless of the platform, which is probably not worth trying to just check the appearance of a single file. If you need a more sophisticated file system looking in addition to this, this might be worth the effort. It seems that pyinotify can be modified to add a subclass of the Notifier class that connects to the asyncio event asyncio (for example, there are already classes for tornado and asyncore ).

For your simple use case, I think your infinite loop approach to polling is fine, but you can also just schedule event loop callbacks if you want:

 def watch_for_file(file_path, interval=1, loop=None): if not loop: loop = asyncio.get_event_loop() if not os.path.exists(file_path): print("{} not found yet.".format(file_path)) loop.call_later(interval, watch_for_file, file_path, interval, loop) else: print("{} found!".format(file_path)) loop.stop() def main(argv=None): cli_parser = make_cli_parser() args = cli_parser.parse_args(argv) loop = asyncio.get_event_loop() loop.call_soon(watch_for_file, args.file_path) loop.run_forever() 

I'm not sure if this is much more elegant than an endless loop.

Edit:

Just for fun, I wrote a solution using pyinotify :

 import pyinotify import asyncio import argparse import os.path class AsyncioNotifier(pyinotify.Notifier): """ Notifier subclass that plugs into the asyncio event loop. """ def __init__(self, watch_manager, loop, callback=None, default_proc_fun=None, read_freq=0, threshold=0, timeout=None): self.loop = loop self.handle_read_callback = callback pyinotify.Notifier.__init__(self, watch_manager, default_proc_fun, read_freq, threshold, timeout) loop.add_reader(self._fd, self.handle_read) def handle_read(self, *args, **kwargs): self.read_events() self.process_events() if self.handle_read_callback is not None: self.handle_read_callback(self) class EventHandler(pyinotify.ProcessEvent): def my_init(self, file=None, loop=None): if not file: raise ValueError("file keyword argument must be provided") self.loop = loop if loop else asyncio.get_event_loop() self.filename = file def process_IN_CREATE(self, event): print("Creating:", event.pathname) if os.path.basename(event.pathname) == self.filename: print("Found it!") self.loop.stop() def make_cli_parser(): cli_parser = argparse.ArgumentParser(description=__doc__) cli_parser.add_argument('file_path') return cli_parser def main(argv=None): cli_parser = make_cli_parser() args = cli_parser.parse_args(argv) loop = asyncio.get_event_loop() # set up pyinotify stuff wm = pyinotify.WatchManager() mask = pyinotify.IN_CREATE # watched events dir_, filename = os.path.split(args.file_path) if not dir_: dir_ = "." wm.add_watch(dir_, mask) handler = EventHandler(file=filename, loop=loop) notifier = pyinotify.AsyncioNotifier(wm, loop, default_proc_fun=handler) loop.run_forever() if __name__ == '__main__': main() 

Answer 2

Butter https://pypi.python.org/pypi/butter supports asyncio support out of the box, BTW.

 import asyncio from butter.inotify import IN_ALL_EVENTS from butter.asyncio.inotify import Inotify_async @asyncio.coroutine def watcher(loop): inotify = Inotify_async(loop=loop) print(inotify) wd = inotify.watch('/tmp', IN_ALL_EVENTS) for i in range(5): event = yield from inotify.get_event() print(event) inotify.ignore(wd) print('done') event = yield from inotify.get_event() print(event) inotify.close() print(inotify) loop = asyncio.get_event_loop() task = loop.create_task(watcher(loop)) loop.run_until_complete(task) 

Answer 3

According to the documentation you can look at file descriptors. This should help you solve this problem with much less code.