我们从Python开源项目中，提取了以下26个代码示例，用于说明如何使用importlib.__import__()。
def test_importlib_import_relative_class_raises(self): assert __package__ with self.assertRaises(ImportError): importlib.__import__('msg.TestMsg', globals=globals(), level=1) # TODO # def test_double_import_uses_cache(self): # # print_importers() # # Verify that files exists and are importable # import std_msgs.msg as std_msgs # # self.assertTrue(std_msgs.Bool is not None) # self.assertTrue(callable(std_msgs.Bool)) # self.assertTrue(std_msgs.Bool._type == 'std_msgs/Bool') # # import std_msgs.msg as std_msgs2 # # self.assertTrue(std_msgs == std_msgs2)
def test_importlib_import_relative_srv(self): # Verify that files exists and are importable test_srvs = importlib.__import__('srv', globals=globals(), level=1) test_msgs = importlib.__import__('msg', globals=globals(), level=1) self.assert_test_service_classes(test_srvs.TestSrv, test_srvs.TestSrvRequest, test_srvs.TestSrvResponse, test_srvs.TestSrvDeps, test_srvs.TestSrvDepsRequest, test_srvs.TestSrvDepsResponse, test_msgs.TestRosMsgDeps, test_msgs.TestRosMsg) # TODO # def test_double_import_uses_cache(self): # # print_importers() # # Verify that files exists and are importable # import std_msgs.msg as std_msgs # # self.assertTrue(std_msgs.Bool is not None) # self.assertTrue(callable(std_msgs.Bool)) # self.assertTrue(std_msgs.Bool._type == 'std_msgs/Bool') # # import std_msgs.msg as std_msgs2 # # self.assertTrue(std_msgs == std_msgs2)
def bench(name, cleanup=lambda: None, *, seconds=1, repeat=3): """Bench the given statement as many times as necessary until total executions take one second.""" stmt = "__import__({!r})".format(name) timer = timeit.Timer(stmt) for x in range(repeat): total_time = 0 count = 0 while total_time < seconds: try: total_time += timer.timeit(1) finally: cleanup() count += 1 else: # One execution too far if total_time > seconds: count -= 1 yield count // seconds
def main(import_): __builtins__.__import__ = import_ benchmarks = (from_cache, builtin_mod, source_using_bytecode, source_wo_bytecode, source_writing_bytecode, decimal_using_bytecode, decimal_writing_bytecode, decimal_wo_bytecode,) seconds = 1 seconds_plural = 's' if seconds > 1 else '' repeat = 3 header = "Measuring imports/second over {} second{}, best out of {}\n" print(header.format(seconds, seconds_plural, repeat)) for benchmark in benchmarks: print(benchmark.__doc__, "[", end=' ') sys.stdout.flush() results = [] for result in benchmark(seconds=seconds, repeat=repeat): results.append(result) print(result, end=' ') sys.stdout.flush() assert not sys.dont_write_bytecode print("]", "best is", format(max(results), ',d'))
def importModule(self, baseName, module): """Load a python module that is a loadable Qudi module. @param string baseName: the module base package (hardware, logic, or gui) @param string module: the python module name inside the base package @return object: the loaded python module """ logger.info('Loading module ".{0}.{1}"'.format(baseName, module)) if not isBase(baseName): raise Exception('You are trying to cheat the ' 'system with some category {0}'.format(baseName)) # load the python module mod = importlib.__import__('{0}.{1}'.format( baseName, module), fromlist=['*']) # print('refcnt:', sys.getrefcount(mod)) return mod
def load(name): try: obj = __import__(name) reload(obj) return obj except: pass try: import importlib obj = importlib.__import__(name) importlib.reload(obj) return obj except: pass
def test_importlib_import_absolute_msg(self): # Verify that files exists and are importable std_msgs = importlib.__import__('std_msgs.msg') std_msgs = std_msgs.msg self.assert_std_message_classes(std_msgs.Bool, std_msgs.Header)
def test_importlib_import_absolute_class_raises(self): with self.assertRaises(ImportError): importlib.__import__('std_msgs.msg.Bool') # BROKEN 3.4 ?
def test_importlib_import_relative_msg(self): # Verify that files exists and are importable test_msgs = importlib.__import__('msg', globals=globals(), level=1) self.assert_test_message_classes(test_msgs.TestMsg, test_msgs.TestMsgDeps, test_msgs.TestRosMsgDeps, test_msgs.TestRosMsg) # BROKEN 3.4 ?
def test_importlib_import_absolute_srv(self): # Verify that files exists and are importable std_srvs = importlib.__import__('std_srvs.srv') std_srvs = std_srvs.srv self.assert_std_service_classes(std_srvs.SetBool, std_srvs.SetBoolRequest, std_srvs.SetBoolResponse)
def test_importlib_import_relative_msg(self): # Verify that files exists and are importable subtest_msgs = importlib.__import__('msg', globals=globals(), level=1) test_msgs = importlib.__import__('test_rosimport.msg') test_msgs = test_msgs.msg self.assert_test_message_classes(subtest_msgs.SubTestMsg, subtest_msgs.SubTestMsgDeps, test_msgs.TestRosMsgDeps, test_msgs.TestRosMsg) # BROKEN 3.4 ?
def test_importlib_import_absolute_srv(self): # Verify that files exists and are importable # __import__ checks sys.modules by itself # but the test is not reflecting anything if we use the already loaded module. if sys.modules.get('std_srvs.srv'): #TODO : EVERYWHERE ! raise unittest.SkipTest("module previously loaded".format('std_srvs.srv')) else: std_srvs = importlib.__import__('std_srvs.srv') std_srvs = std_srvs.srv self.assert_std_service_classes(std_srvs.SetBool, std_srvs.SetBoolRequest, std_srvs.SetBoolResponse)
def import_(*args, **kwargs): """Delegate to allow for injecting different implementations of import.""" if using___import__: return __import__(*args, **kwargs) else: return importlib.__import__(*args, **kwargs)
def importlib_only(fxn): """Decorator to skip a test if using __builtins__.__import__.""" return unittest.skipIf(using___import__, "importlib-specific test")(fxn)
def generate_cf_dynamo_schema(target): dynamo_connection = DynamoDBConnection() class FakeClient(object): def create_table(self, *args, **kwargs): write_schema_to_yaml(target, **kwargs) return {} client = FakeClient() dynamo_connection.client = client class FakeDynamo(object): def list_tables(self): return [] def create_table(self, *args): result = dynamo_connection.create_table(*args) def describe_table(self, *args): StatusStruct = namedtuple('Status', 'status') return StatusStruct(status='ACTIVE') dynamo = FakeDynamo() engine = Engine() engine.dynamo = dynamo sys.path = ['{}/app/models'.format(target)] + sys.path modelModules = glob.glob('{}/app/models'.format(target) + '/*.py') models = [basename(f)[:-3] for f in modelModules if isfile(f)] for modelName in models: if modelName != '__init__': engine.register(getattr(importlib.__import__(modelName), modelName)) engine.create_schema()
def rundb(target): """ Start running a local DynamoDB instance. :param target: :return: """ load_env(target) os.environ['AWS_REGION'] = 'us-west-2' shared_db = './dynamo_db/shared-local-instance.db' if os.path.exists(shared_db): os.remove(shared_db) dynamo_command = ['java', '-Djava.library.path={}/dynamo_db/DynamoDBLocal_lib'.format(CWD), '-jar', '{}/dynamo_db/DynamoDBLocal.jar'.format(CWD), '-sharedDb', '-dbPath', './dynamo_db'] try: dynamo_process = subprocess.Popen(dynamo_command, stdin=subprocess.PIPE, stderr=subprocess.STDOUT) except Exception as e: pass try: ''' Connect to DynamoDB and register and create tables for application models. ''' engine = Engine() engine.connect(os.environ['AWS_REGION'], host='localhost', port=8000, access_key='anything', secret_key='anything', is_secure=False) # load models sys.path = ['./app/models'] + sys.path modelModules = glob.glob('./app/models' + "/*.py") models = [basename(f)[:-3] for f in modelModules if isfile(f)] for modelName in models: if modelName != '__init__': engine.register(getattr(__import__(modelName), modelName)) engine.create_schema() tables = [table for table in engine.dynamo.list_tables()] print("This engine has the following tables " + str(tables)) for table in tables: engine.dynamo.describe_table(table) except Exception as e: # IF anything goes wrong, then we self-destruct. dynamo_process.kill() raise e # Wait for process to finish. dynamo_process.wait()
