Source code for jina.importer

import importlib
import os
import sys
import warnings
from types import SimpleNamespace
from typing import Optional

from jina import __resources_path__

IMPORTED = SimpleNamespace()
IMPORTED.executors = False
IMPORTED.schema_executors = {}

[docs]class ImportExtensions: """ A context manager for wrapping extension import and fallback. It guides the user to pip install correct package by looking up extra-requirements.txt. :param required: set to True if you want to raise the ModuleNotFound error :param logger: when not given, built-in warnings.warn will be used :param help_text: the help text followed after :param pkg_name: the package name to find in extra_requirements.txt, when not given the ModuleNotFound exec_val will be used as the best guess """ def __init__( self, required: bool, logger=None, help_text: Optional[str] = None, pkg_name: Optional[str] = None, verbose: bool = True, ): self._required = required self._tags = [] self._help_text = help_text self._logger = logger self._pkg_name = pkg_name self._verbose = verbose def __enter__(self): return self def __exit__(self, exc_type, exc_val, traceback): if exc_type == ModuleNotFoundError: missing_module = self._pkg_name or with open(os.path.join(__resources_path__, 'extra-requirements.txt')) as fp: for v in fp: if ( v.strip() and not v.startswith('#') and v.startswith(missing_module) and ':' in v ): missing_module, install_tags = v.split(':') self._tags.append(missing_module) self._tags.extend(vv.strip() for vv in install_tags.split(',')) break if self._tags: from jina.helper import colored req_msg = colored('fallback to default behavior', color='yellow') if self._required: req_msg = colored('and it is required', color='red') err_msg = f'''Python package "{colored(missing_module, attrs='bold')}" is not installed, {req_msg}. You are trying to use a feature not enabled by your current Jina installation.''' avail_tags = ' '.join( colored(f'[{tag}]', attrs='bold') for tag in self._tags ) err_msg += ( f'\n\nTo enable this feature, use {colored("pip install jina[TAG]", attrs="bold")}, ' f'where {colored("[TAG]", attrs="bold")} is one of {avail_tags}.\n' ) else: err_msg = f'{exc_val.msg}' if self._required: if self._verbose: if self._logger: self._logger.critical(err_msg) if self._help_text: self._logger.error(self._help_text) else: warnings.warn(err_msg, RuntimeWarning, stacklevel=2) raise exc_val else: if self._verbose: if self._logger: self._logger.warning(err_msg) if self._help_text: else: warnings.warn(err_msg, RuntimeWarning, stacklevel=2) return True # suppress the error
def _path_import(absolute_path: str): import importlib.util try: # I dont want to trust user path based on directory structure, "user_module", period default_spec_name = 'user_module' user_module_name = os.path.splitext(os.path.basename(absolute_path))[0] if user_module_name == '__init__': # __init__ can not be used as a module name spec_name = default_spec_name elif user_module_name not in sys.modules: spec_name = user_module_name else: warnings.warn( f''' {user_module_name} shadows one of built-in Python module name. It is imported as `{default_spec_name}.{user_module_name}` Affects: - Either, change your code from using `from {user_module_name} import ...` to `from {default_spec_name}.{user_module_name} import ...` - Or, rename {user_module_name} to another name ''' ) spec_name = f'{default_spec_name}.{user_module_name}' spec = importlib.util.spec_from_file_location(spec_name, absolute_path) module = importlib.util.module_from_spec(spec) sys.modules[spec_name] = module spec.loader.exec_module(module) except Exception as ex: raise ImportError(f'can not import module from {absolute_path}') from ex
[docs]class PathImporter: """The class to import modules from paths."""
[docs] @staticmethod def add_modules(*paths): """ Import modules from paths. :param paths: Paths of the modules. """ # assume paths are Python module names not_python_module_paths = [] for path in paths: if not os.path.isfile(path): try: importlib.import_module(path) except: not_python_module_paths.append(path) else: not_python_module_paths.append(path) # try again, but assume they are file paths instead of module names from jina.jaml.helper import complete_path for m in not_python_module_paths: _path_import(complete_path(m))