"""A re-implementation of zipimport to use importlib. XXX Differences with 2.x version: + _zip_directory_cache uses instances of zipfile.ZipInfo as values instead of tuples. + zipimporter.prefix does not necessarily end in a path separator. + ZipImportError is raised if the argument to zipimporter.__init__ does not contain a path to a zip file; 2.x version allows for opening a any file but raises errors later on during usage. + ZipImportError is raised even if None is passed in to zipimport.__init__. """ import importlib import contextlib import datetime import imp import os import time import zipfile # XXX _zip_directory_cache issues: # * Import lock prevents concurrency issues during importation use, but does # not make any guarantees for non-import uses. # * Uses zipfile.ZipInfo instances for values of file paths (C version uses # tuples). # * Prevents any dynamic update to the zip file from being detected. _zip_directory_cache = {} class ZipImportError(ImportError): pass class zipimporter(object): _handler = importlib.handle_code _init_module = importlib.init_module def __init__(self, archivepath): """Open the specified zip file. Meant to be used from sys.path_hooks, meaning that if something other than a zip file is passed in then ZipImportError is raised. """ path = archivepath while path and path != os.sep: if zipfile.is_zipfile(path): break path = os.path.split(path)[0] else: raise ZipImportError("%s is not a path to a zip file" % archivepath) self.archive = os.path.abspath(path) # Path to zip file. self.prefix = archivepath[len(self.archive)+1:] # Package directory. if not self.archive in _zip_directory_cache: with contextlib.closing(zipfile.ZipFile(path)) as zip_file: zip_info_list = zip_file.infolist() zip_info_dict = dict((info.filename, info) for info in zip_info_list) _zip_directory_cache[self.archive] = zip_info_dict self._files = _zip_directory_cache[self.archive] def __repr__(self): if self.prefix: path = os.path.join(self.archive, self.prefix) else: path = self.archive return '' % path def _check_paths(self, tail, pkg=False): """Check if the module (or package) is contained within the package represented by the importer.""" source_suffixes = importlib.suffix_list(imp.PY_SOURCE) bytecode_suffixes = importlib.suffix_list(imp.PY_COMPILED) source, bytecode = None, None files_in_zip = _zip_directory_cache[self.archive] if pkg: tail = os.path.join(tail, pkg) for suffix in source_suffixes: path = os.path.join(self.prefix, tail + suffix) if path in files_in_zip: source = path break for suffix in bytecode_suffixes: path = os.path.join(self.prefix, tail + suffix) if path in files_in_zip: bytecode = path break if source is not None or bytecode is not None: return source, bytecode else: return None def find_module(self, fullname, path=None): """Check if the specified module is contained within the zip file, returning self if it is or None if it is not.""" tail = fullname.rpartition('.')[2] # Look for a package. results = self._check_paths(tail, '__init__') if results is not None: return self results = self._check_paths(tail) if results is not None: return self return None def _handle_code(self, name): """Get the code object, the path used, and whether the name points to a package.""" tail = name.rpartition('.')[2] info = self._check_paths(tail, '__init__') if info: pkg = True else: info = self._check_paths(tail) if info: pkg = False else: raise ZipImportError('%s is not known' % name) if info[0]: source_path = os.path.join(self.archive, info[0]) else: source_path = None if info[1]: bytecode_path = os.path.join(self.archive, info[1]) else: bytecode_path = None return self._handler(name, source_path, bytecode_path) + (pkg,) def get_code(self, fullname): """Return the code object for the module, raising ZipImportError if the module is not found.""" return self._handle_code(fullname)[0] def get_data(self, pathname): """Return the data (raw source or bytecode) for the specified module, or raise IOError if the path does not exist. All files are opened in binary mode, regardless of whether they are source or bytecode files. """ if pathname.startswith(self.archive): pathname = pathname[len(self.archive)+1:] # Cover path separator. with contextlib.closing(zipfile.ZipFile(self.archive)) as zip_: try: return zip_.open(pathname, 'r').read() except KeyError: raise IOError('%s does not exist in the zip file %s' % (pathname, self.archive)) def get_source(self, fullname): """Get the source code for the specified module, raising ZipImportError if the module does not exist, or None if only bytecode exists.""" tail = fullname.rpartition('.')[2] paths = self._check_paths(tail, '__init__') if paths: if paths[0]: source_path = paths[0] else: return None else: paths = self._check_paths(tail) if paths: if paths[0]: source_path = paths[0] else: return None else: raise ZipImportError("%s is not known" % fullname) with contextlib.closing(zipfile.ZipFile(self.archive)) as zip_: return zip_.open(source_path, 'U').read() def is_package(self, fullname): """Return True if the module name represents a package, False if it does not.""" tail = fullname.rpartition('.')[2] if self._check_paths(tail, '__init__'): return True elif self._check_paths(tail): return False else: raise ZipImportError("%s is not known" % fullname) def mod_time(self, name): """Return the last modification time of the module's source code from the epoch (based on the local time).""" tail = name.rpartition('.')[2] info = self._check_paths(tail, '__init__') if not info: info = self._check_paths(tail) if not info[0]: raise ZipImportError("no source for %s" % name) file_info = _zip_directory_cache[self.archive][info[0]] file_mtime = datetime.datetime(*file_info.date_time) return int(time.mktime(file_mtime.timetuple())) def get_bytecode(self, name): """Get the bytecode for the module.""" tail = name.rpartition('.')[2] info = self._check_paths(tail, '__init__') if not info: info = self._check_paths(tail) if info and info[1]: with contextlib.closing(zipfile.ZipFile(self.archive)) as zip_: bytecode = zip_.open(info[1], 'r').read() return bytecode[:4], importlib._r_long(bytecode[4:8]), bytecode[8:] elif info: return None else: raise ZipImportError('%s is not known' % name) def write_bytecode(self, name, timestamp, data): """Return False as zip files are never modified.""" # XXX Worth implementing ability to write out to zip files? return False def load_module(self, fullname): """Load the specified module, returning the module or raising ZipImportError if the module could not be found.""" code_object, path, is_pkg = self._handle_code(fullname) return self._init_module(code_object, fullname, path, is_pkg)