Coverage for mindsdb / integrations / libs / ml_handler_process / update_process.py: 16%

1import importlib 

2 

3from mindsdb.interfaces.storage.model_fs import HandlerStorage, ModelStorage 

4 

5 

6def update_process(args: dict, integration_id: int, module_path: str, model_id: int) -> None: 

7 module = importlib.import_module(module_path) 

8 

9 if module.import_error is not None: 

10 raise module.import_error 

11 

12 result = None 

13 

14 if hasattr(module.Handler, "upgate"): 

15 engine_storage = HandlerStorage(integration_id) 

16 model_storage = ModelStorage(model_id) 

17 try: 

18 result = module.Handler(engine_storage=engine_storage, model_storage=model_storage).upgate(args=args) 

19 except NotImplementedError: 

20 return None 

21 

22 return result