bottle 中一个有意思的上下文管理器

监测文本改动，如果有改动则使主线程退出 bottle.py

这个类有enter和exit这两个dunder方法，实现了上下文管理器协议。在进入这个上下文管理器的时候，启动这个线程，退出时等待线程结束，且忽略了KeyboardInterrupt异常，因为exit返回True之外的值时，with中的异常才会向上冒泡。在run方法中在for module in list(sys.modules.values()):...这个for循环中拿到所有module文件的modify time。然后在之后的while循环中，监测文件改动，如果有改动调用thread.interrupt_main()，在主线程（bottle所在线程）中raise，KeyboardInterrupt异常。

if sys.version_info.major > 2:
    import _thread as thread
else:
    import thread

class FileCheckerThread(threading.Thread):
    """ Interrupt main-thread as soon as a changed module file is detected,
        the lockfile gets deleted or gets too old. """

    def __init__(self, lockfile, interval):
        threading.Thread.__init__(self)
        self.daemon = True
        self.lockfile, self.interval = lockfile, interval
        #: Is one of 'reload', 'error' or 'exit'
        self.status = None

    def run(self):
        exists = os.path.exists
        mtime = lambda p: os.stat(p).st_mtime
        files = dict()

        for module in list(sys.modules.values()):
            path = getattr(module, '__file__', '')
            if path[-4:] in ('.pyo', '.pyc'): path = path[:-1]
            if path and exists(path): files[path] = mtime(path)  # 拿到所有导入模块文件的modify time

        while not self.status:
            if not exists(self.lockfile)\
            or mtime(self.lockfile) < time.time() - self.interval - 5:
                self.status = 'error'
                thread.interrupt_main()
            for path, lmtime in list(files.items()):
                if not exists(path) or mtime(path) > lmtime:  # 如果文件发生改动，
                    self.status = 'reload'
                    thread.interrupt_main()  # raise 一个 KeyboardInterrupt exception in 主线程
                    break 
            time.sleep(self.interval)

    def __enter__(self):
        self.start()

    def __exit__(self, exc_type, *_):
        if not self.status: self.status = 'exit'  # silent exit
        self.join()
        return exc_type is not None and issubclass(exc_type, KeyboardInterrupt)

使用

import time
import threading
import _thread as thread

class CheckerThread(threading.Thread):

    def __init__(self, interval):
        threading.Thread.__init__(self)
        self.daemon = True
        self.interval = interval

    def run(self):
        time.sleep(self.interval)
        thread.interrupt_main()  # raise 一个 KeyboardInterrupt exception in 主线程

    def __enter__(self):
        self.start()

    def __exit__(self, exc_type, *_):
        self.join()
        return exc_type is not None and issubclass(exc_type, KeyboardInterrupt)

if __name__ == "__main__":
    fc = CheckerThread(5)
    with fc:
        while True:
            print('haha')
            time.sleep(1)

    print('end')

"""输出
haha
haha
haha
haha
haha
end
"""

kagxin / blog

bottle 中一个有意思的上下文管理器 #17

监测文本改动，如果有改动则使主线程退出 bottle.py

使用