如何监控Python文件的变化?

发布于 2024-09-26 00:21:00 字数 143 浏览 3 评论 0原文

如果代码发生更改,我想重新启动我的 Python Web 应用程序。但是可能有大量文件可以更改,因为导入模块中的文件可能会更改...

如何从导入的包/模块中获取实际的文件名?

如何高效检测修改过的Python文件?有图书馆可以做到这一点吗?

I want to restart my Python web application, if code gets changed. But there could be a large number of files that could be changed, since files in imported modules could change ...

How to get the actual file names from imported packages / modules?

How can modified Python files be detected efficiently? Is there a library to do that?

如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。

扫码二维码加入Web技术交流群

发布评论

需要 登录 才能够评论, 你可以免费 注册 一个本站的账号。

评论(5

(り薆情海 2024-10-03 00:21:00

无耻的插头。我还正在努力做到这一点 http://github.com/gorakhargosh/watchdog

HTH。

Shameless plug. There's also http://github.com/gorakhargosh/watchdog that I'm working on to do exactly this.

HTH.

风筝有风,海豚有海 2024-10-03 00:21:00

gamin 是另一个不太特定于 Linux 的选项。

gamin is another option which is slightly less Linux-specific.

遗忘曾经 2024-10-03 00:21:00

我不确定您将如何在您的情况下实现“重新加载应用程序”操作;使用内置的 reload 重新加载已更改的模块可能不会解决问题。

但就检测是否发生变化而言,以下是一种处理方法。

  • 大多数 python 模块都有一个 __file__ 属性。
  • 所有加载的模块都存储在sys.modules中。
  • 我们可以以一定的间隔遍历 sys.modules,并依次在磁盘上查找每个模块的更改。

有时,__file__ 指向一个 .pyc文件而不是 .py 文件,因此您可能必须删除尾随的 c。有时 .pyc 文件存在,但 .py 不存在;在一个强大的系统中,你必须允许这一点。

执行此操作的代码概念证明(不可靠):

_module_timestamps = {}
_checking = False

def run_checker():
    global _checking
    _checking = True
    while _checking:
        for name, module in sys.modules.iteritems():
            if hasattr(module, '__file__'):
                filename = module.__file__
                if filename.endswith('.pyc'):
                    filename = filename[:-1]
                mtime = os.stat(filename).st_mtime
                if name not in _module_timestamps:
                    _module_timestamps[name] = mtime
                else:
                    if mtime > _module_timestamps[name]:
                        do_reload(name)
            else:
                'module %r has no file attribute' % (name,)
        time.sleep(1)

def do_reload(modname):
    print 'I would reload now, because of %r' % (modname,)

check_thread = threading.Thread(target=run_checker)
check_thread.daemon = True
check_thread.start()

try:
    while 1:
        time.sleep(0.1)
except KeyboardInterrupt:
    print '\nexiting...'

I'm not sure how you would implement the 'reload application' operation in your circumstance; reloading a changed module with the reload built-in probably won't cut it.

But as far as detecting whether or not there was a change, the following would be one way to approach it.

  • Most python modules have a __file__ attribute.
  • All loaded modules are stored in sys.modules.
  • We can walk through sys.modules at some interval, and look for changes on disk for each module in turn

Sometimes __file__ points to a .pyc file instead of a .py file, so you might have to chop off the trailing c. Sometimes a .pyc file exists but a .py doesn't exist; in a robust system you'd have to allow for this.

A proof of concept of code to do this (not robust):

_module_timestamps = {}
_checking = False

def run_checker():
    global _checking
    _checking = True
    while _checking:
        for name, module in sys.modules.iteritems():
            if hasattr(module, '__file__'):
                filename = module.__file__
                if filename.endswith('.pyc'):
                    filename = filename[:-1]
                mtime = os.stat(filename).st_mtime
                if name not in _module_timestamps:
                    _module_timestamps[name] = mtime
                else:
                    if mtime > _module_timestamps[name]:
                        do_reload(name)
            else:
                'module %r has no file attribute' % (name,)
        time.sleep(1)

def do_reload(modname):
    print 'I would reload now, because of %r' % (modname,)

check_thread = threading.Thread(target=run_checker)
check_thread.daemon = True
check_thread.start()

try:
    while 1:
        time.sleep(0.1)
except KeyboardInterrupt:
    print '\nexiting...'
月棠 2024-10-03 00:21:00

下面是如何使用 pyinotify(即在 Linux 上)实现这一点的示例。

from importlib import import_module

class RestartingLauncher:

    def __init__(self, module_name, start_function, stop_function, path="."):
        self._module_name = module_name
        self._filename = '%s.py' % module_name
        self._start_function = start_function
        self._stop_function = stop_function
        self._path = path
        self._setup()

    def _setup(self):
        import pyinotify
        self._wm = pyinotify.WatchManager()

        self._notifier = pyinotify.ThreadedNotifier(
                self._wm, self._on_file_modified)
        self._notifier.start()

        # We monitor the directory (instead of just the file) because
        # otherwise inotify gets confused by editors such a Vim.
        flags = pyinotify.EventsCodes.OP_FLAGS['IN_MODIFY']
        wdd = self._wm.add_watch(self._path, flags)

    def _on_file_modified(self, event):
        if event.name == self._filename:
            print "File modification detected. Restarting application..."
            self._reload_request = True
            getattr(self._module, self._stop_function)()

    def run(self):
        self._module = import_module(self._module_name)

        self._reload_request = True
        while self._reload_request:
            self._reload_request = False
            reload(self._module)
            getattr(self._module, self._start_function)()

        print 'Bye!'
        self._notifier.stop()

def launch_app(module_name, start_func, stop_func):
    try:
        import pyinotify
    except ImportError:
        print 'Pyinotify not found. Launching app anyway...'
        m = import_module(self._module_name)
        getattr(m, start_func)()
    else:
        RestartingLauncher(module_name, start_func, stop_func).run()

if __name__ == '__main__':
    launch_app('example', 'main', 'force_exit')

launch_app 调用中的参数是文件名(不带“.py”)、开始执行的函数和以某种方式停止执行的函数。

这是一个可以使用以前的代码(重新)启动的“应用程序”的愚蠢示例:

run = True

def main():
    print 'in...'
    while run: pass
    print 'out'

def force_exit():
    global run
    run = False

在您想要使用它的典型应用程序中,您可能会有某种主循环。这是一个更真实的示例,适用于基于 GLib/GTK+ 的应用程序:

from gi.repository import GLib

GLib.threads_init()
loop = GLib.MainLoop()

def main():
    print "running..."
    loop.run()

def force_exit():
    print "stopping..."
    loop.quit()

相同的概念适用于大多数其他循环(Clutter、Qt 等)。

监视多个代码文件(即属于应用程序一部分的所有文件)和错误恢复(例如,打印异常并在空闲循环中等待,直到代码修复,然后再次启动)留给读者作为练习:) 。

注意:此答案中的所有代码均根据 ISC 许可证(除了知识共享)发布。

Here's an example of how this could be implemented using pyinotify (ie., on Linux).

from importlib import import_module

class RestartingLauncher:

    def __init__(self, module_name, start_function, stop_function, path="."):
        self._module_name = module_name
        self._filename = '%s.py' % module_name
        self._start_function = start_function
        self._stop_function = stop_function
        self._path = path
        self._setup()

    def _setup(self):
        import pyinotify
        self._wm = pyinotify.WatchManager()

        self._notifier = pyinotify.ThreadedNotifier(
                self._wm, self._on_file_modified)
        self._notifier.start()

        # We monitor the directory (instead of just the file) because
        # otherwise inotify gets confused by editors such a Vim.
        flags = pyinotify.EventsCodes.OP_FLAGS['IN_MODIFY']
        wdd = self._wm.add_watch(self._path, flags)

    def _on_file_modified(self, event):
        if event.name == self._filename:
            print "File modification detected. Restarting application..."
            self._reload_request = True
            getattr(self._module, self._stop_function)()

    def run(self):
        self._module = import_module(self._module_name)

        self._reload_request = True
        while self._reload_request:
            self._reload_request = False
            reload(self._module)
            getattr(self._module, self._start_function)()

        print 'Bye!'
        self._notifier.stop()

def launch_app(module_name, start_func, stop_func):
    try:
        import pyinotify
    except ImportError:
        print 'Pyinotify not found. Launching app anyway...'
        m = import_module(self._module_name)
        getattr(m, start_func)()
    else:
        RestartingLauncher(module_name, start_func, stop_func).run()

if __name__ == '__main__':
    launch_app('example', 'main', 'force_exit')

The parameters in the launch_app call are the filename (without the ".py"), the function to start execution and a function that somehow stops the execution.

Here's a stupid example of an "app" that could be (re-)launched using the previous code:

run = True

def main():
    print 'in...'
    while run: pass
    print 'out'

def force_exit():
    global run
    run = False

In a typical application where you'd want to use this, you'd probably have a main loop of some sort. Here's a more real example, for a GLib/GTK+ based application:

from gi.repository import GLib

GLib.threads_init()
loop = GLib.MainLoop()

def main():
    print "running..."
    loop.run()

def force_exit():
    print "stopping..."
    loop.quit()

The same concept works for most other loops (Clutter, Qt, etc).

Monitoring several code files (ie. all files that are part of the application) and error resilience (eg. printing exceptions and waiting in an idle loop until the code is fixed, then launching it again) are left as exercises for the reader :).

Note: All code in this answer is released under the ISC License (in addition to Creative Commons).

£噩梦荏苒 2024-10-03 00:21:00

这是特定于操作系统的。对于 Linux,有 inotify,请参见 http://github.com/rvoicilas/inotify-tools/< /a>

This is operating system specific. For Linux, there is inotify, see e.g. http://github.com/rvoicilas/inotify-tools/

~没有更多了~
我们使用 Cookies 和其他技术来定制您的体验包括您的登录状态等。通过阅读我们的 隐私政策 了解更多相关信息。 单击 接受 或继续使用网站,即表示您同意使用 Cookies 和您的相关数据。
原文