-
-
Save jezdez/3103339 to your computer and use it in GitHub Desktop.
PEP 302 import hook to reload modules on changes
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# apkg/__init__.py | |
# Empty. Nothing to see here. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# apkg/hello.py | |
from __future__ import print_function | |
print("HELLO WORLD") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
"""PEP 302 import hook to reload modules on changes. | |
Implements an import hook using a finder and a loader class to load modules | |
from a filesystem path using the existing imp module. The modification time of | |
all Python source files are recorded. The modifiation times can then be used | |
later to generate a list of modified modules, unload modified modules or unload | |
all modules if any are modified. | |
To use, create an instance of ReloadingFinder and append it to sys.meta_path. | |
The search path used by ReloadingFinder can altered by passing an iterable of | |
path names to the instance when created. | |
Tested with Python 2.6, 2.7 and 3.2. | |
""" | |
from __future__ import print_function | |
from collections import namedtuple | |
import imp | |
import os | |
import sys | |
# containers for common data sets | |
Location = namedtuple('Location', 'fobject pathname description') | |
Source = namedtuple('Source', 'pathname mtime') | |
class ReloadingLoader(object): | |
"""Import hook loader. | |
Used by the ReloadingFinder class. Implemented using the imp module. | |
""" | |
def __init__(self, source): | |
self.source = source | |
def load_module(self, fullname): | |
"Import hook protocol." | |
try: | |
# print('ReloadingLoader loading "{0}"'.format(fullname)) | |
return imp.load_module(fullname, *self.source) | |
except Exception as err: | |
print(err) | |
raise | |
finally: | |
if hasattr(self.source.fobject, 'close'): | |
self.source.fobject.close() | |
class ReloadingFinder(object): | |
"""Import hook loader. | |
Finds modules using the imp module. Records the modification time for | |
module source files. The default search path for modules is the current | |
directory. This can be over ridden by passing a iterable of paths name when | |
creating the instance. | |
Instances have the following properties. | |
- loaded | |
- modified | |
With the names of modules that have been loaded and modified since loaded. | |
Deleting this properties unloads the modules. Instances also have an | |
ismodified property that is True if there are modified modules. | |
""" | |
def __init__(self, path=(os.path.abspath(os.curdir),)): | |
self._default = list(path) | |
self._mtimes = {} | |
@property | |
def ismodified(self): | |
for _ in self._iter_modified(): | |
return True | |
return False | |
@property | |
def loaded(self): | |
return list(self._mtimes.keys()) | |
@loaded.deleter | |
def loaded(self): | |
for fullname in self.loaded: | |
del sys.modules[fullname] | |
del self._mtimes[fullname] | |
@property | |
def modified(self): | |
return [module for module in self._iter_modified()] | |
@modified.deleter | |
def modified(self): | |
for fullname in self.modified: | |
del sys.modules[fullname] | |
del self._mtimes[fullname] | |
def _choose_path(self, pkgpath): | |
if pkgpath is None: | |
return self._default | |
return pkgpath | |
def _find_location(self, module, path): | |
# print('ReloadingFinder finding "{0}" in "{1}"'.format(module, path)) | |
fobject, pathname, description = imp.find_module(module, path) | |
location = Location(fobject, pathname, description) | |
return location | |
def _iter_modified(self): | |
for fullname in self._mtimes: | |
location = self._mtimes[fullname] | |
stat = os.stat(location.pathname) | |
if stat.st_mtime > location.mtime: | |
yield fullname | |
def _module_name(self, fullname): | |
hierarchy = fullname.rsplit('.', 1) | |
return hierarchy[-1] | |
def _record_mtime(self, fullname, location): | |
stat = os.stat(location.pathname) | |
source = Source(location.pathname, stat.st_mtime) | |
self._mtimes[fullname] = source | |
def find_module(self, fullname, pkgpath=None): | |
"Import hook protocol." | |
try: | |
module = self._module_name(fullname) | |
path = self._choose_path(pkgpath) | |
location = self._find_location(module, path) | |
self._record_mtime(fullname, location) | |
return ReloadingLoader(location) | |
except Exception as err: | |
print(err) | |
raise | |
# create and register the import hook | |
finder = ReloadingFinder() | |
sys.meta_path.append(finder) | |
if __name__ == '__main__': | |
# import 'apkg.hello' for the first time. | |
# will print 'HELLO WORLD' to standard out. | |
import apkg.hello | |
# import 'apkg.hello' for a second time. | |
# already imported so no side effect. | |
import apkg.hello | |
# no modules have been changed. | |
# two modules have been loaded | |
if finder.ismodified: | |
print('Modified,', finder.modified) | |
print('Loaded,', len(finder.loaded)) | |
# unload modules that have been imported | |
# use 'finder.modified' to unload just modified | |
del finder.loaded | |
# now no modules have been loaded | |
print('Loaded,', len(finder.loaded)) | |
# import 'apkg.hello' for the third time. | |
# it will again print 'HELLO WORLD" | |
import apkg.hello |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment