Coverage for mindsdb / integrations / libs / ml_handler_process / create_engine_process.py: 21%

1import importlib 

2 

3from pandas import DataFrame 

4 

5from mindsdb.interfaces.storage.model_fs import HandlerStorage 

6 

7 

8def create_engine_process(connection_args: dict, integration_id: int, module_path: str) -> DataFrame: 

9 module = importlib.import_module(module_path) 

10 

11 if module.import_error is not None: 

12 raise module.import_error 

13 

14 result = None 

15 

16 if hasattr(module.Handler, "create_engine"): 

17 engine_storage = HandlerStorage(integration_id) 

18 try: 

19 result = module.Handler(engine_storage=engine_storage, model_storage=None).create_engine( 

20 connection_args=connection_args 

21 ) 

22 except NotImplementedError: 

23 return None 

24 

25 return result