我正在尝试使用 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 session 中，我发出

touch testfile

结束循环。

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

最佳答案

好吧，有更好的、特定于平台的方式可以在文件创建时收到通知。 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()

关于python - 使用 asyncio 监视文件，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/26414052/