注意使用asyncio查找文件

Question

我正在尝试使用Python 的 asyncio 库来寻找一种好的方法来监控文件的出现。这是我目前想到的：

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

在另一个终端会话中，我输入：

touch testfile

结束循环。

除了使用无限循环和 yield from asyncio.sleep()，是否还有更优雅的解决方案？

- gotgenes

---

您IP地址为143.198.54.68，由于运营成本限制，当前对于免费用户的使用频率限制为每个IP每72小时10次对话，如需解除限制，请点击左下角设置图标按钮（手机用户先点击左上角菜单按钮）。 - Gerrat

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

- dano

---

更新一下，我提交了一个补丁来添加“AsyncioNotifier”到“pyinotify”，并且已经被接受。因此，未来的版本应该已经内置了这个支持。 - dano

Answer 2

顺便提一下，Butterhttps://pypi.python.org/pypi/butter已经默认支持asyncio了。

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)

- Andrew Svetlov

---

"asyncio.coroutine" 拼写错误。此外，根据 asyncio 文档，不应该实例化 "Task"，而应该使用 "async()" 或 "BaseEventLoop.create_task()"。 - gotgenes

好的，首选的方式是使用loop.create_task(coro)，但只使用Task(coro, loop=loop)也可以。总之已经修复了。 - Andrew Svetlov

目前这个不兼容Python 3.7。 - lig

抱歉，什么不兼容？ - Andrew Svetlov

Answer 3

Butter非常棒。另一个替代方案是minotaur，它类似，但只实现了inotify。

async def main():
    with Inotify(blocking=False) as n:
        n.add_watch('.', Mask.CREATE | Mask.DELETE | Mask.MOVE)
        async for evt in n:
            print(evt)

- scaramanga