Python Examples of importlib.import_module (2024)

def enable_plugin(self, plugin): """Enable a ingester plugin for use parsing design documents. :params plugin: - A string naming a class object denoting the ingester plugin to be enabled """ if plugin is None or plugin == '': self.log.error("Cannot have an empty plugin string.") try: (module, x, classname) = plugin.rpartition('.') if module == '': raise Exception() mod = importlib.import_module(module) klass = getattr(mod, classname) self.registered_plugin = klass() except Exception as ex: self.logger.error( "Could not enable plugin %s - %s" % (plugin, str(ex))) if self.registered_plugin is None: self.logger.error("Could not enable at least one plugin") raise Exception("Could not enable at least one plugin") 

def complaints(): username = request.authorization.username extractor = Extractor.query.filter_by(username=username).first() department = extractor.first_department() request_json = request.json added_rows = 0 updated_rows = 0 complaint_class = getattr(importlib.import_module("comport.data.models"), "CitizenComplaint{}".format(department.short_name)) for incident in request_json['data']: added = complaint_class.add_or_update_incident(department, incident) if added is True: added_rows += 1 elif added is False: updated_rows += 1 extractor.next_month = None extractor.next_year = None extractor.save() return json.dumps({"added": added_rows, "updated": updated_rows}) 

def register_all(): # NOTE(sh8121att) - Import all versioned objects so # they are available via RPC. Any new object definitions # need to be added here. importlib.import_module('drydock_provisioner.objects.network') importlib.import_module('drydock_provisioner.objects.node') importlib.import_module('drydock_provisioner.objects.hostprofile') importlib.import_module('drydock_provisioner.objects.hwprofile') importlib.import_module('drydock_provisioner.objects.site') importlib.import_module('drydock_provisioner.objects.promenade') importlib.import_module('drydock_provisioner.objects.rack') importlib.import_module('drydock_provisioner.objects.bootaction') importlib.import_module('drydock_provisioner.objects.task') importlib.import_module('drydock_provisioner.objects.builddata') importlib.import_module('drydock_provisioner.objects.validation')# Utility class for calculating inheritance 

def load_project(project_name): # load project and check that it looks okay try: importlib.import_module(project_name) except ImportError as e: try: #TODO: relative module imports in a projects/Project will fail for some reason importlib.import_module("projects.%s" % project_name) except ImportError as e: log.error("Failed to import project %s", project_name, exc_info=1) sys.exit(1) if len(_registered) != 1: log.error("Project must register itself using alf.register(). " "%d projects registered, expecting 1.", len(_registered)) sys.exit(1) project_cls = _registered.pop() if not issubclass(project_cls, Fuzzer): raise TypeError("Expecting a Fuzzer, not '%s'" % type(project_cls)) return project_cls 

def __init__(self, name=None, # Network name. Used to select TensorFlow name and variable scopes. func=None, # Fully qualified name of the underlying network construction function. **static_kwargs): # Keyword arguments to be passed in to the network construction function. self._init_fields() self.name = name self.static_kwargs = dict(static_kwargs) # Init build func. module, self._build_func_name = import_module(func) self._build_module_src = inspect.getsource(module) self._build_func = find_obj_in_module(module, self._build_func_name) # Init graph. self._init_graph() self.reset_vars() 

def add_step(self,module_name_and_params, extra_args): config=module_name_and_params.split() module_name=config[0] params=config[1:] # collect extra arguments from command line meant for this particular module if extra_args is not None: for _name, _value in extra_args.__dict__.items(): if _name.startswith(module_name): _modname,_argname=_name.split(".",1) # for example lemmatizer_mod.gpu params.append("--"+_argname) params.append(str(_value)) mod=importlib.import_module(module_name) step_in=self.q_out self.q_out=Queue(self.max_q_size) #new pipeline end args=mod.argparser.parse_args(params) process=Process(target=mod.launch,args=(args,step_in,self.q_out)) process.daemon=True process.start() self.processes.append(process) 

def get_model_function(model_type): """ Get tensorflow function of the model to be applied to the input tensor. For instance "unet.softmax_unet" will return the softmax_unet function in the "unet.py" submodule of the current module (spleeter.model). Params: - model_type: str the relative module path to the model function. Returns: A tensorflow function to be applied to the input tensor to get the multitrack output. """ relative_path_to_module = '.'.join(model_type.split('.')[:-1]) model_name = model_type.split('.')[-1] main_module = '.'.join((__name__, 'functions')) path_to_module = f'{main_module}.{relative_path_to_module}' module = importlib.import_module(path_to_module) model_function = getattr(module, model_name) return model_function 

def _import_channel(channel_alias): """ helper to import channels aliases from string paths. raises an AttributeError if a channel can't be found by it's alias """ try: channel_path = settings.NOTIFICATIONS_CHANNELS[channel_alias] except KeyError: raise AttributeError( '"%s" is not a valid delivery channel alias. ' 'Check your applications settings for NOTIFICATIONS_CHANNELS' % channel_alias ) package, attr = channel_path.rsplit('.', 1) return getattr(importlib.import_module(package), attr) 

def use_of_force(): username = request.authorization.username extractor = Extractor.query.filter_by(username=username).first() department = extractor.first_department() request_json = request.json added_rows = 0 updated_rows = 0 uof_class = getattr(importlib.import_module("comport.data.models"), "UseOfForceIncident{}".format(department.short_name)) for incident in request_json['data']: added = uof_class.add_or_update_incident(department, incident) if added is True: added_rows += 1 elif added is False: updated_rows += 1 extractor.next_month = None extractor.next_year = None extractor.save() return json.dumps({"added": added_rows, "updated": updated_rows}) 

def officer_involved_shooting(): username = request.authorization.username extractor = Extractor.query.filter_by(username=username).first() department = extractor.first_department() request_json = request.json added_rows = 0 updated_rows = 0 ois_class = getattr(importlib.import_module("comport.data.models"), "OfficerInvolvedShooting{}".format(department.short_name)) for incident in request_json['data']: added = ois_class.add_or_update_incident(department, incident) if added is True: added_rows += 1 elif added is False: updated_rows += 1 extractor.next_month = None extractor.next_year = None extractor.save() return json.dumps({"added": added_rows, "updated": updated_rows}) 

def pursuits(): username = request.authorization.username extractor = Extractor.query.filter_by(username=username).first() department = extractor.first_department() request_json = request.json added_rows = 0 updated_rows = 0 pursuit_class = getattr(importlib.import_module("comport.data.models"), "Pursuit{}".format(department.short_name)) for incident in request_json['data']: added = pursuit_class.add_or_update_incident(department, incident) if added is True: added_rows += 1 elif added is False: updated_rows += 1 extractor.next_month = None extractor.next_year = None extractor.save() return json.dumps({"added": added_rows, "updated": updated_rows}) 

def assaults(): username = request.authorization.username extractor = Extractor.query.filter_by(username=username).first() department = extractor.first_department() request_json = request.json added_rows = 0 updated_rows = 0 assaults_class = getattr(importlib.import_module("comport.data.models"), "AssaultOnOfficer{}".format(department.short_name)) for incident in request_json['data']: added = assaults_class.add_or_update_incident(department, incident) if added is True: added_rows += 1 elif added is False: updated_rows += 1 extractor.next_month = None extractor.next_year = None extractor.save() return json.dumps({"added": added_rows, "updated": updated_rows}) 

def _make_context(): ''' Return context dict for a shell session. ''' # generate a list of all the incident classes class_lookup = [ {"UseOfForceIncident": ["IMPD", "BPD", "LMPD"]}, {"CitizenComplaint": ["IMPD", "BPD"]}, {"OfficerInvolvedShooting": ["IMPD", "BPD"]}, {"AssaultOnOfficer": ["IMPD"]} ] incident_classes = {} for guide in class_lookup: prefix, departments = guide.popitem() for name in departments: class_name = prefix + name incident_classes[class_name] = getattr(importlib.import_module("comport.data.models"), class_name) context = {'app': app, 'db': db, 'User': User, 'Department': Department, 'Extractor': Extractor, 'IncidentsUpdated': IncidentsUpdated, 'JSONTestClient': JSONTestClient} # update the context with the incident classes context.update(incident_classes) return context 

def test_csv_filtered_by_dept(self, testapp): # create a department department1 = Department.create(name="IM Police Department", short_name="IMPD", load_defaults=False) department2 = Department.create(name="B Police Department", short_name="BPD", load_defaults=False) incidentclass1 = getattr(importlib.import_module("comport.data.models"), "UseOfForceIncident{}".format(department1.short_name)) incidentclass2 = getattr(importlib.import_module("comport.data.models"), "UseOfForceIncident{}".format(department2.short_name)) incidentclass1.create(opaque_id="123ABC", department_id=department1.id) incidentclass2.create(opaque_id="123XYZ", department_id=department2.id) response1 = testapp.get("/department/{}/uof.csv".format(department1.id)) response2 = testapp.get("/department/{}/uof.csv".format(department2.id)) incidents1 = list(csv.DictReader(io.StringIO(response1.text))) incidents2 = list(csv.DictReader(io.StringIO(response2.text))) assert len(incidents1) == 1 and len(incidents2) == 1 assert incidents1[0]['id'] == '123ABC' and incidents2[0]['id'] == '123XYZ' 

def import_usr_dir(usr_dir): """Import module at usr_dir, if provided.""" if not usr_dir: return if usr_dir == INTERNAL_USR_DIR_PACKAGE: # The package has been installed with pip under this name for Cloud ML # Engine so just import it. importlib.import_module(INTERNAL_USR_DIR_PACKAGE) return dir_path = os.path.abspath(os.path.expanduser(usr_dir).rstrip("/")) containing_dir, module_name = os.path.split(dir_path) tf.logging.info("Importing user module %s from path %s", module_name, containing_dir) sys.path.insert(0, containing_dir) importlib.import_module(module_name) sys.path.pop(0) 

def get_anonymizers(app_config): anonymizers_module = app_config.name + '.anonymizers' mod = importlib.import_module(anonymizers_module) anonymizers = [] for value in mod.__dict__.values(): if value is Anonymizer: continue is_anonymizer = False try: if issubclass(value, Anonymizer) and value.model is not None: is_anonymizer = True except TypeError: continue if is_anonymizer: anonymizers.append(value) anonymizers.sort(key=lambda c: c.order) return anonymizers 

def get_module_chain(names): r"""Attempt to load a module from an ordered list of module names until one succeeds. Module names may be given fully as: EXOSIMS.OpticalSystem.Nemati or as wildcards like: EXOSIMS.*.Nemati Wildcards, if given, must match only one module. """ for name in names: # replace name by its expansion if needed if '*' in name: name = wildcard_expand(name) try: full_module = importlib.import_module(name) #module_name = package #source = package return full_module except ImportError: continue return None 

def __new__(cls, driver_wrapper=None): """Instantiate android or ios page object from base page object depending on driver configuration Base, Android and iOS page objects must be defined with following structure: FOLDER/base/MODULE_NAME.py class BasePAGE_OBJECT_NAME(MobilePageObject) FOLDER/android/MODULE_NAME.py class AndroidPAGE_OBJECT_NAME(BasePAGE_OBJECT_NAME) FOLDER/ios/MODULE_NAME.py class IosPAGE_OBJECT_NAME(BasePAGE_OBJECT_NAME) :param driver_wrapper: driver wrapper instance :returns: android or ios page object instance """ if cls.__name__.startswith('Base'): __driver_wrapper = driver_wrapper if driver_wrapper else DriverWrappersPool.get_default_wrapper() __os_name = 'ios' if __driver_wrapper.is_ios_test() else 'android' __class_name = cls.__name__.replace('Base', __os_name.capitalize()) try: return getattr(importlib.import_module(cls.__module__), __class_name)(__driver_wrapper) except AttributeError: __module_name = cls.__module__.replace('.base.', '.{}.'.format(__os_name)) return getattr(importlib.import_module(__module_name), __class_name)(__driver_wrapper) else: return super(MobilePageObject, cls).__new__(cls) 

def __init__(self, pywren_config, storage_backend): self.pywren_config = pywren_config self.backend = storage_backend self._created_cobjects_n = itertools.count() try: module_location = 'pywren_ibm_cloud.storage.backends.{}'.format(self.backend) sb_module = importlib.import_module(module_location) storage_config = self.pywren_config[self.backend] storage_config['user_agent'] = 'pywren-ibm-cloud/{}'.format(__version__) StorageBackend = getattr(sb_module, 'StorageBackend') self.storage_handler = StorageBackend(storage_config) except Exception as e: raise NotImplementedError("An exception was produced trying to create the " "'{}' storage backend: {}".format(self.backend, e)) 

def __init__(self, storage_config, executor_id=None): self.config = storage_config self.backend = self.config['backend'] self.bucket = self.config['bucket'] self.executor_id = executor_id self._created_cobjects_n = itertools.count() try: module_location = 'pywren_ibm_cloud.storage.backends.{}'.format(self.backend) sb_module = importlib.import_module(module_location) StorageBackend = getattr(sb_module, 'StorageBackend') self.storage_handler = StorageBackend(self.config[self.backend], bucket=self.bucket, executor_id=self.executor_id) except Exception as e: raise NotImplementedError("An exception was produced trying to create the " "'{}' storage backend: {}".format(self.backend, e)) 

def load(self, func_name=None, _type="auto"): FUNCTIONS = get_functions() if _type == "auto": _type = "cython" if self.is_compiled else "python" if func_name not in FUNCTIONS: raise Exception("Function %s not found: %s" % (func_name, FUNCTIONS.keys())) func = FUNCTIONS[func_name] if _type not in func: raise Exception("Module not available in %s." % _type) func = func[_type] # either provide a function or a string to the module (used for cython) if not callable(func): module = importlib.import_module(func) func = getattr(module, func_name) return func 

def run(request): """ This runs the model that was picked. """ env = None try: action = request.POST[ACTION] except KeyError: action = None session_id = int(request.session['session_id']) # Load entry_point model_name = request.POST[MODEL] model = ABMModel.objects.get(name=model_name) entry_point = model.module plot_type = model.plot_type importlib.import_module(entry_point[0:-4]) questions = model.params.all() # Take actions on a running model if action: env = running_model(request, action, entry_point, questions, session_id) # Run a model for the first time else: env = model_first_run(request, action, entry_point, questions, session_id, plot_type) site_hdr = get_hdr() text_box, image_bytes = env.user.text_output, env.plot() image = base64.b64encode(image_bytes.getvalue()).decode() template_data = {HEADER: site_hdr, 'text0': text_box[0], 'image': image, 'text1': text_box[1], 'model': model} return render(request, 'run.html', template_data) 

def get_env(model_nm, pa): module = importlib.import_module('models.{model_nm}'.format(model_nm=model_nm.lower())) Env = getattr(module, '{}Env'.format(model_nm)) return Env(model_nm=model_nm, props=pa) 

def load_module(self, fullname): print('load_module {}'.format(fullname)) if fullname in sys.modules: return sys.modules[fullname] # 先从 sys.meta_path 中删除自定义的 finder # 防止下面执行 import_module 的时候再次触发此 finder # 从而出现递归调用的问题 finder = sys.meta_path.pop(0) # 导入 module module = importlib.import_module(fullname) module_hook(fullname, module) sys.meta_path.insert(0, finder) return module 

def load_module(self, fullname): print('load_module {}'.format(fullname)) if fullname in sys.modules: return sys.modules[fullname] # 先从 sys.meta_path 中删除自定义的 finder # 防止下面执行 import_module 的时候再次触发此 finder # 从而出现递归调用的问题 finder = sys.meta_path.pop(0) # 导入 module module = importlib.import_module(fullname) module_hook(fullname, module) sys.meta_path.insert(0, finder) return module 

def load_module(self, fullname): print('load_module {}'.format(fullname)) if fullname in sys.modules: return sys.modules[fullname] # 先从 sys.meta_path 中删除自定义的 finder # 防止下面执行 import_module 的时候再次触发此 finder # 从而出现递归调用的问题 finder = sys.meta_path.pop(0) # 导入 module module = importlib.import_module(fullname) module_hook(fullname, module) sys.meta_path.insert(0, finder) return module 

def _import_modules(module_names): imported_modules = [] for module_name in module_names: module = importlib.import_module(module_name) if hasattr(module, 'list_opts'): imported_modules.append(module) return imported_modules 

def try_import(mod_name): """Attempt importing module and suppress failure of doing this.""" try: return importlib.import_module(mod_name) except ImportError: pass 

def load_module(name): """ Import or reload tutorial module as needed. """ target = 'cherrypy.tutorial.' + name if target in sys.modules: module = importlib.reload(sys.modules[target]) else: module = importlib.import_module(target) return module 

def import_module(module_or_obj_name): parts = module_or_obj_name.split('.') parts[0] = {'np': 'numpy', 'tf': 'tensorflow'}.get(parts[0], parts[0]) for i in range(len(parts), 0, -1): try: module = importlib.import_module('.'.join(parts[:i])) relative_obj_name = '.'.join(parts[i:]) return module, relative_obj_name except ImportError: pass raise ImportError(module_or_obj_name)