На этом шаге мы рассмотрим второй подход к кастомизации импортирования.
Второй подход к кастомизации импортирования - написать хук, который прикрепляется напрямую к переменной sys.path, распознавая некоторые шаблоны наименования каталогов. Добавьте приведенный ниже класс и поддерживающие функции в urlimport.py:
# urlimport.py import sys import importlib.abc import imp from urllib.request import urlopen from urllib.error import HTTPError, URLError from html.parser import HTMLParser # Отладка import logging log = logging.getLogger(__name__) # Получение ссылок из переданного URL def _get_links(url): class LinkParser(HTMLParser): def handle_starttag(self, tag, attrs): if tag == 'a': attrs = dict(attrs) links.add(attrs.get('href').rstrip('/')) links = set() try: log.debug('Getting links from %s' % url) u = urlopen(url) parser = LinkParser() parser.feed(u.read().decode('utf-8')) except Exception as e: log.debug('Could not get links. %s', e) log.debug('links: %r', links) return links class UrlMetaFinder(importlib.abc.MetaPathFinder): def __init__(self, baseurl): self._baseurl = baseurl self._links = {} self._loaders = {baseurl: UrlModuleLoader(baseurl)} def find_module(self, fullname, path=None): log.debug('find_module: fullname=%r, path=%r', fullname, path) if path is None: baseurl = self._baseurl else: if not path[0].startswith(self._baseurl): return None baseurl = path[0] parts = fullname.split('.') basename = parts[-1] log.debug('find_module: baseurl=%r, basename=%r', baseurl, basename) # Проверка кеша ссылок if basename not in self._links: self._links[baseurl] = _get_links(baseurl) # Проверка того, что это не пакет if basename in self._links[baseurl]: log.debug('find_module: trying package %r', fullname) fullurl = self._baseurl + '/' + basename # Попытка загрузить пакет (которая обращается к __init__.py) loader = UrlPackageLoader(fullurl) try: loader.load_module(fullname) self._links[fullurl] = _get_links(fullurl) self._loaders[fullurl] = UrlModuleLoader(fullurl) log.debug('find_module: package %r loaded', fullname) except ImportError as e: log.debug('find_module: package failed. %s', e) loader = None return loader # Обычный модуль filename = basename + '.py' if filename in self._links[baseurl]: log.debug('find_module: module %r found', fullname) return self._loaders[baseurl] else: log.debug('find_module: module %r not found', fullname) return None def invalidate_caches(self): log.debug('invalidating link cache') self._links.clear() # Загрузчик модуля из URL class UrlModuleLoader(importlib.abc.SourceLoader): def __init__(self, baseurl): self._baseurl = baseurl self._source_cache = {} def module_repr(self, module): return '<urlmodule %r from %r>' % (module.__name__, module.__file__) # Требуемый метод def load_module(self, fullname): code = self.get_code(fullname) mod = sys.modules.setdefault(fullname, imp.new_module(fullname)) mod.__file__ = self.get_filename(fullname) mod.__loader__ = self mod.__package__ = fullname.rpartition('.')[0] exec(code, mod.__dict__) return mod # Необязательные расширения def get_code(self, fullname): src = self.get_source(fullname) return compile(src, self.get_filename(fullname), 'exec') def get_data(self, path): pass def get_filename(self, fullname): return self._baseurl + '/' + fullname.split('.')[-1] + '.py' def get_source(self, fullname): filename = self.get_filename(fullname) log.debug('loader: reading %r', filename) if filename in self._source_cache: log.debug('loader: cached %r', filename) return self._source_cache[filename] try: u = urlopen(filename) source = u.read().decode('utf-8') log.debug('loader: %r loaded', filename) self._source_cache[filename] = source return source except (HTTPError, URLError) as e: log.debug('loader: %r failed. %s', filename, e) raise ImportError("Can't load %s" % filename) def is_package(self, fullname): return False # Загрузчик пакета из URL class UrlPackageLoader(UrlModuleLoader): def load_module(self, fullname): mod = super().load_module(fullname) mod.__path__ = [self._baseurl] mod.__package__ = fullname def get_filename(self, fullname): return self._baseurl + '/' + '__init__.py' def is_package(self, fullname): return True # Вспомогательные функции для инсталляции/деинсталляции загрузчика _installed_meta_cache = {} def install_meta(address): if address not in _installed_meta_cache: finder = UrlMetaFinder(address) _installed_meta_cache[address] = finder sys.meta_path.append(finder) log.debug('%r installed on sys.meta_path', finder) def remove_meta(address): if address in _installed_meta_cache: finder = _installed_meta_cache.pop(address) sys.meta_path.remove(finder) log.debug('%r removed from sys.meta_path', finder) # ======== Начало добавления ================ # Класс для поиска пути в URL class UrlPathFinder(importlib.abc.PathEntryFinder): def __init__(self, baseurl): self._links = None self._loader = UrlModuleLoader(baseurl) self._baseurl = baseurl def find_loader(self, fullname): log.debug('find_loader: %r', fullname) parts = fullname.split('.') basename = parts[-1] # Проверка кеша ссылок if self._links is None: self._links = [] # See discussion self._links = _get_links(self._baseurl) # Проверить, пакет ли это if basename in self._links: log.debug('find_loader: trying package %r', fullname) fullurl = self._baseurl + '/' + basename # Попытка загрузить пакет (которая обращается к __init__.py) loader = UrlPackageLoader(fullurl) try: loader.load_module(fullname) log.debug('find_loader: package %r loaded', fullname) except ImportError as e: log.debug('find_loader: %r is a namespace package', fullname) loader = None return (loader, [fullurl]) # Обычный модуль filename = basename + '.py' if filename in self._links: log.debug('find_loader: module %r found', fullname) return (self._loader, []) else: log.debug('find_loader: module %r not found', fullname) return (None, []) def invalidate_caches(self): log.debug('invalidating link cache') self._links = None # Проверка пути на предмет того, выглядит ли он как URL _url_path_cache = {} def handle_url(path): if path.startswith(('http://', 'https://')): log.debug('Handle path? %s. [Yes]', path) if path in _url_path_cache: finder = _url_path_cache[path] else: finder = UrlPathFinder(path) _url_path_cache[path] = finder return finder else: log.debug('Handle path? %s. [No]', path) def install_path_hook(): sys.path_hooks.append(handle_url) sys.path_importer_cache.clear() log.debug('Installing handle_url') def remove_path_hook(): sys.path_hooks.remove(handle_url) sys.path_importer_cache.clear() log.debug('Removing handle_url')
Чтобы использовать этот основанный на пути поисковик, просто добавьте URL'ы в sys.path. Например:
>>> # Первоначальный импорт не удается >>> import fib Traceback (most recent call last): File "<stdin>", line 1, in <module> ImportError: No module named 'fib' >>> # Установка хука пути >>> import urlimport >>> urlimport.install_path_hook() >>> # Импортирование все еще не удается (не в пути) >>> import fib Traceback (most recent call last): File "<stdin>", line 1, in <modu1e> ImportError: No module named 'fib' >>> # Добавляем запись в sys.path и наблюдаем, как все отлично работает >>> import sys >>> sys.path.append('http://localhost:15000') >>> import fib I'm fib >>> import grok.blah I'm grok.__init__ I'm grok.blah >>> grok.blah.__file__ 'http://localhost:15000/grok/blah.py' >>>
Ключ к этому последнему примеру заключается в функции handle_url(), которая добавлена в переменную sys.path_hooks. Когда обрабатываются записи sys.path, вызываются функции из sys.path_hooks. Если какая-либо из этих функций возвращает объект поисковика, этот поисковик используется для попытки загрузки модулей для данной записи в sys.path.
Стоит отметить, что удаленно импортируемые модули работают точно так же, как и любые другие. Например:
>>> fib <urlmodule 'fib' from 'http://localhost:15000/fib.py'> >>> fib.__name__ 'fib' >>> fib.__file__ 'http://localhost:15dee/fib.py' >>> import inspect >>> print(inspect.getsource(fib)) # fib.py print("I'm fib") def fib(n): if n < 2: return 1 else: return fib(n-1) + fib(n-2) >>>
На следующем шаге мы обсудим этот вопрос.