如何监视Python文件的更改？

Question

如果代码被更改，我想重新启动我的Python Web应用程序。但可能会有大量的文件可以更改，因为进口模块中的文件可能会更改...

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

如何有效检测修改后的Python文件？有没有图书馆可以做到这一点？

Answer 1

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

Answer 2

gamin是另一种选择，它比Linux特定的稍差。

Answer 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...' 

Answer 4

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

HTH。

Answer 5

下面是如何使用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 

在典型应用中，你会想利用这一点，你可能有某种主循环。这里有一个更真实例如，能说会道/ 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() 

同样的概念也适用于大多数其他循环（杂波，Qt的，等等）。

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

注意：此答案中的所有代码均在ISC许可证下发布（除知识共享之外）。