观察带有asyncio的文件

Question

我试图找出使用Python's asyncio library来观察文件外观的好方法。这是我想出迄今：

#!/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() 

我救以此为watch_for_file.py，并且可以与

python3 watch_for_file.py testfile 

运行在另一个shell会话，我发出

touch testfile 

结束循环。

有没有比使用这个无限循环更优雅的解决方案和yield from asyncio.sleep()？

Answer 1

那么，有更好的，特定于平台的方式来创建文件时得到通知。 Gerrat在他的评论中将Windows链接到了一个，并且pyinotify可以用于Linux。这些特定于平台的方法可能会被插入到asyncio中，但是最终你会写一大堆代码来使其以独立于平台的方式工作，这可能不值得花费精力来检查单个文件。但是，如果您需要更复杂的文件系统，那么这可能值得追求。例如，看起来pyinotify可以调整为将Notifier类的子类添加到asyncio事件循环中（例如，已有类tornado和asyncore）。

为您简单的用例，我觉得你的无限循环的方式来投票是好的，但你也可以只安排回调与事件循环，如果你想：

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() 

我不知道虽然这比无限循环更优雅。

编辑：

只是为了好玩，我写了使用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

黄油https://pypi.python.org/pypi/butter有ASYNCIO开箱即用的支持，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

根据documentation，您现在可以观察文件描述符。这应该可以帮助你用更少的代码来解决这个问题。