【问题标题】:How to monitor Python files for changes?如何监视 Python 文件的更改?
【发布时间】:2011-04-21 06:22:27
【问题描述】:

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

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

如何有效地检测修改过的 Python 文件?有图书馆可以做吗?

【问题讨论】:

标签: python file notifications monitoring


【解决方案1】:

这是一个如何使用 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 许可下发布(除了知识共享)。

【讨论】:

    【解决方案2】:

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

    HTH。

    【讨论】:

    • 我刚刚使用 Watchdog 编写了一个可爱的小脚本:zolomon.com/wp/?p=382 - 我说这是一个真正的魅力!
    • 有没有使用看门狗重启现有进程的例子,如果它的任何源文件发生变化?
    【解决方案3】:

    我不确定您将如何在您的情况下实现“重新加载应用程序”操作;使用内置的 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...'
    

    【讨论】:

      【解决方案4】:

      gamin 是另一个选项,它与 Linux 的相关性稍差。

      【讨论】:

        【解决方案5】:

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

        【讨论】:

          猜你喜欢
          • 1970-01-01
          • 2011-06-18
          • 1970-01-01
          • 1970-01-01
          • 1970-01-01
          • 1970-01-01
          • 1970-01-01
          • 1970-01-01
          • 1970-01-01
          相关资源
          最近更新 更多