blob: 1a02e69436fafe00f748707bc19088a2cbf2d769 [file] [log] [blame]
from typing import List, Callable, Dict, Optional, Any, Union
import builtins
import importlib
from torch.serialization import _load
import pickle
import torch
import _compat_pickle # type: ignore
import types
import os.path
from ._importlib import _normalize_line_endings, _resolve_name, _sanity_check, _calc___package__, \
_normalize_path
from ._mock_zipreader import MockZipReader
class PackageImporter:
"""Importers allow you to load code written to packages by PackageExporter.
Code is loaded in a hermetic way, using files from the package
rather than the normal python import system. This allows
for the packaging of PyTorch model code and data so that it can be run
on a server or used in the future for transfer learning.
The importer for packages ensures that code in the module can only be loaded from
within the package, except for modules explicitly listed as external during export.
The file `extern_modules` in the zip archive lists all the modules that a package externally depends on.
This prevents "implicit" dependencies where the package runs locally because it is importing
a locally-installed package, but then fails when the package is copied to another machine.
"""
modules : Dict[str, Optional[types.ModuleType]]
"""The dictionary of already loaded modules from this package, equivalent to `sys.modules` but
local to this importer.
"""
def __init__(self, filename: str, module_allowed: Callable[[str], bool] = lambda module_name: True):
"""Open `filename` for importing. This checks that the imported package only requires modules
allowed by `module_allowed`
Args:
filename (str): archive to load. Can also be a directory of the unzipped files in the archive
for easy debugging and editing.
module_allowed (Callable[[str], bool], optional): A method to determine if a externally provided module
should be allowed. Can be used to ensure packages loaded do not depend on modules that the server
does not support. Defaults to allowing anything.
Raises:
ImportError: If the package will use a disallowed module.
"""
self.filename = filename
self.zip_reader : Any
if not os.path.isdir(self.filename):
self.zip_reader = torch._C.PyTorchFileReader(self.filename)
else:
self.zip_reader = MockZipReader(self.filename)
self.root = _PackageNode(None)
self.modules = {}
self.extern_modules = self._read_extern()
for extern_module in self.extern_modules:
if not module_allowed(extern_module):
raise ImportError(f"package '{filename}' needs the external module '{extern_module}' "
f"but that module has been disallowed")
self._add_extern(extern_module)
for filename in self.zip_reader.get_all_records():
self._add_file(filename)
self.patched_builtins = builtins.__dict__.copy()
self.patched_builtins['__import__'] = self.__import__
# allow pickles from archive using `import resources`
self.modules['resources'] = self # type: ignore
# used for torch.serialization._load
self.Unpickler = lambda *args, **kwargs: _UnpicklerWrapper(self, *args, **kwargs)
def import_module(self, name: str, package=None):
"""Load a module from the package if it hasn't already been loaded, and then return
the module. Modules are loaded locally
to the importer and will appear in `self.modules` rather than `sys.modules`
Args:
name (str): Fully qualified name of the module to load.
package ([type], optional): Unused, but present to match the signature of importlib.import_module. Defaults to None.
Returns:
types.ModuleType: the (possibly already) loaded module.
"""
return self._gcd_import(name)
def load_binary(self, package: str, resource: str) -> bytes:
"""Load raw bytes.
Args:
package (str): The name of module package (e.g. "my_package.my_subpackage")
resource (str): The unique name for the resource.
Returns:
bytes: The loaded data.
"""
path = self._zipfile_path(package, resource)
return self.zip_reader.get_record(path)
def load_text(self, package: str, resource: str, encoding: str = 'utf-8', errors: str = 'strict') -> str:
"""Load a string.
Args:
package (str): The name of module package (e.g. "my_package.my_subpackage")
resource (str): The unique name for the resource.
encoding (str, optional): Passed to `decode`. Defaults to 'utf-8'.
errors (str, optional): Passed to `decode`. Defaults to 'strict'.
Returns:
str: The loaded text.
"""
data = self.load_binary(package, resource)
return data.decode(encoding, errors)
def load_pickle(self, package: str, resource: str, map_location=None) -> Any:
"""Unpickles the resource from the package, loading any modules that are needed to construct the objects
using :meth:`import_module`
Args:
package (str): The name of module package (e.g. "my_package.my_subpackage")
resource (str): The unique name for the resource.
map_location: Passed to `torch.load` to determine how tensors are mapped to devices. Defaults to None.
Returns:
Any: the unpickled object.
"""
pickle_file = self._zipfile_path(package, resource)
return _load(self.zip_reader, map_location, self, pickle_file=pickle_file)
def _read_extern(self):
return self.zip_reader.get_record('extern_modules').decode('utf-8').splitlines(keepends=False)
def _make_module(self, name: str, filename: Optional[str], is_package: bool):
spec = importlib.machinery.ModuleSpec(name, self, is_package=is_package) # type: ignore
module = importlib.util.module_from_spec(spec)
self.modules[name] = module
ns = module.__dict__
ns['__spec__'] = spec
ns['__loader__'] = self
ns['__file__'] = filename
ns['__cached__'] = None
ns['__builtins__'] = self.patched_builtins
if filename is not None:
code = self._compile_source(filename)
exec(code, ns)
return module
def _load_module(self, name: str):
cur : _PathNode = self.root
for atom in name.split('.'):
if not isinstance(cur, _PackageNode) or atom not in cur.children:
raise ModuleNotFoundError(
f'No module named "{name}" in self-contained archive "{self.filename}"'
f' and the module is also not in the list of allowed external modules: {self.extern_modules}')
cur = cur.children[atom]
if isinstance(cur, _ExternNode):
module = self.modules[name] = importlib.import_module(name)
return module
return self._make_module(name, cur.source_file, isinstance(cur, _PackageNode)) # type: ignore
def _compile_source(self, fullpath):
source = self.zip_reader.get_record(fullpath)
source = _normalize_line_endings(source)
return compile(source, fullpath, 'exec', dont_inherit=True)
# note: named `get_source` so that linecache can find the source
# when this is the __loader__ of a module.
def get_source(self, module_name) -> str:
module = self.import_module(module_name)
return self.zip_reader.get_record(module.__file__).decode('utf-8')
# note: copied from cpython's import code, with call to create module replaced with _make_module
def _do_find_and_load(self, name):
path = None
parent = name.rpartition('.')[0]
if parent:
if parent not in self.modules:
self._gcd_import(parent)
# Crazy side-effects!
if name in self.modules:
return self.modules[name]
parent_module = self.modules[parent]
try:
path = parent_module.__path__ # type: ignore
except AttributeError:
msg = (_ERR_MSG + '; {!r} is not a package').format(name, parent)
raise ModuleNotFoundError(msg, name=name) from None
module = self._load_module(name)
if parent:
# Set the module as an attribute on its parent.
parent_module = self.modules[parent]
if parent_module.__loader__ is self: # type: ignore
setattr(parent_module, name.rpartition('.')[2], module)
return module
# note: copied from cpython's import code
def _find_and_load(self, name):
module = self.modules.get(name, _NEEDS_LOADING)
if module is _NEEDS_LOADING:
return self._do_find_and_load(name)
if module is None:
message = ('import of {} halted; '
'None in sys.modules'.format(name))
raise ModuleNotFoundError(message, name=name)
return module
def _gcd_import(self, name, package=None, level=0):
"""Import and return the module based on its name, the package the call is
being made from, and the level adjustment.
This function represents the greatest common denominator of functionality
between import_module and __import__. This includes setting __package__ if
the loader did not.
"""
_sanity_check(name, package, level)
if level > 0:
name = _resolve_name(name, package, level)
return self._find_and_load(name)
# note: copied from cpython's import code
def _handle_fromlist(self, module, fromlist, *, recursive=False):
"""Figure out what __import__ should return.
The import_ parameter is a callable which takes the name of module to
import. It is required to decouple the function from assuming importlib's
import implementation is desired.
"""
# The hell that is fromlist ...
# If a package was imported, try to import stuff from fromlist.
if hasattr(module, '__path__'):
for x in fromlist:
if not isinstance(x, str):
if recursive:
where = module.__name__ + '.__all__'
else:
where = "``from list''"
raise TypeError(f"Item in {where} must be str, "
f"not {type(x).__name__}")
elif x == '*':
if not recursive and hasattr(module, '__all__'):
self._handle_fromlist(module, module.__all__,
recursive=True)
elif not hasattr(module, x):
from_name = '{}.{}'.format(module.__name__, x)
try:
self._gcd_import(from_name)
except ModuleNotFoundError as exc:
# Backwards-compatibility dictates we ignore failed
# imports triggered by fromlist for modules that don't
# exist.
if (exc.name == from_name and
self.modules.get(from_name, _NEEDS_LOADING) is not None):
continue
raise
return module
def __import__(self, name, globals=None, locals=None, fromlist=(), level=0):
if level == 0:
module = self._gcd_import(name)
else:
globals_ = globals if globals is not None else {}
package = _calc___package__(globals_)
module = self._gcd_import(name, package, level)
if not fromlist:
# Return up to the first dot in 'name'. This is complicated by the fact
# that 'name' may be relative.
if level == 0:
return self._gcd_import(name.partition('.')[0])
elif not name:
return module
else:
# Figure out where to slice the module's name up to the first dot
# in 'name'.
cut_off = len(name) - len(name.partition('.')[0])
# Slice end needs to be positive to alleviate need to special-case
# when ``'.' not in name``.
return self.modules[module.__name__[:len(module.__name__) - cut_off]]
else:
return self._handle_fromlist(module, fromlist)
def _get_package(self, package):
"""Take a package name or module object and return the module.
If a name, the module is imported. If the passed or imported module
object is not a package, raise an exception.
"""
if hasattr(package, '__spec__'):
if package.__spec__.submodule_search_locations is None:
raise TypeError('{!r} is not a package'.format(
package.__spec__.name))
else:
return package
else:
module = self.import_module(package)
if module.__spec__.submodule_search_locations is None:
raise TypeError('{!r} is not a package'.format(package))
else:
return module
def _zipfile_path(self, package, resource):
package = self._get_package(package)
resource = _normalize_path(resource)
assert package.__loader__ is self
return f"{package.__name__.replace('.', '/')}/{resource}"
def _get_or_create_package(self, atoms: List[str]) -> 'Union[_PackageNode, _ExternNode]':
cur = self.root
for i, atom in enumerate(atoms):
node = cur.children.get(atom, None)
if node is None:
node = cur.children[atom] = _PackageNode(None)
if isinstance(node, _ExternNode):
return node
if isinstance(node, _ModuleNode):
name = ".".join(atoms[:i])
raise ImportError(f'inconsistent module structure. module {name} is not a package, but has submodules')
assert isinstance(node, _PackageNode)
cur = node
return cur
def _add_file(self, filename: str):
*prefix, last = filename.split('/')
package = self._get_or_create_package(prefix)
if isinstance(package, _ExternNode):
raise ImportError(f'inconsistent module structure. package contains a module file {filename}'
f' that is a subpackage of a module marked external.')
if last == '__init__.py':
package.source_file = filename
elif last.endswith('.py'):
package_name = last[:-len('.py')]
package.children[package_name] = _ModuleNode(filename)
def _add_extern(self, extern_name: str):
*prefix, last = extern_name.split('.')
package = self._get_or_create_package(prefix)
if isinstance(package, _ExternNode):
return # the shorter extern covers this extern case
package.children[last] = _ExternNode()
_NEEDS_LOADING = object()
_ERR_MSG_PREFIX = 'No module named '
_ERR_MSG = _ERR_MSG_PREFIX + '{!r}'
class _UnpicklerWrapper(pickle._Unpickler): # type: ignore
def __init__(self, importer, *args, **kwargs):
super().__init__(*args, **kwargs)
self._importer = importer
def find_class(self, module, name):
# Subclasses may override this.
if self.proto < 3 and self.fix_imports:
if (module, name) in _compat_pickle.NAME_MAPPING:
module, name = _compat_pickle.NAME_MAPPING[(module, name)]
elif module in _compat_pickle.IMPORT_MAPPING:
module = _compat_pickle.IMPORT_MAPPING[module]
mod = self._importer.import_module(module)
return getattr(mod, name)
class _PathNode:
pass
class _PackageNode(_PathNode):
def __init__(self, source_file: Optional[str]):
self.source_file = source_file
self.children : Dict[str, _PathNode] = {}
class _ModuleNode(_PathNode):
__slots__ = ['source_file']
def __init__(self, source_file: str):
self.source_file = source_file
class _ExternNode(_PathNode):
pass