Coverage for mlos_bench/mlos_bench/storage/storage_factory.py: 30%

1# 

2# Copyright (c) Microsoft Corporation. 

3# Licensed under the MIT License. 

4# 

5""" 

6Factory method to create a new Storage instance from configs. 

7""" 

8 

9from typing import Any, Optional, List, Dict 

10 

11from mlos_bench.config.schemas import ConfigSchema 

12from mlos_bench.services.config_persistence import ConfigPersistenceService 

13from mlos_bench.storage.base_storage import Storage 

14 

15 

16def from_config(config_file: str, 

17 global_configs: Optional[List[str]] = None, 

18 **kwargs: Any) -> Storage: 

19 """ 

20 Create a new storage object from JSON5 config file. 

21 

22 Parameters 

23 ---------- 

24 config_file : str 

25 JSON5 config file to load. 

26 global_configs : Optional[List[str]] 

27 An optional list of config files with global parameters. 

28 kwargs : dict 

29 Additional configuration parameters. 

30 

31 Returns 

32 ------- 

33 storage : Storage 

34 A new storage object. 

35 """ 

36 config_path: List[str] = kwargs.get("config_path", []) 

37 config_loader = ConfigPersistenceService({"config_path": config_path}) 

38 global_config = {} 

39 for fname in (global_configs or []): 

40 config = config_loader.load_config(fname, ConfigSchema.GLOBALS) 

41 global_config.update(config) 

42 config_path += config.get("config_path", []) 

43 config_loader = ConfigPersistenceService({"config_path": config_path}) 

44 global_config.update(kwargs) 

45 

46 class_config = config_loader.load_config(config_file, ConfigSchema.STORAGE) 

47 assert isinstance(class_config, Dict) 

48 

49 ret = config_loader.build_storage( 

50 service=config_loader, 

51 config=class_config, 

52 global_config=global_config, 

53 ) 

54 assert isinstance(ret, Storage) 

55 return ret