觀察帶有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，您現在可以觀察文件描述符。這應該可以幫助你用更少的代碼來解決這個問題。