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

1# 

2# Copyright (c) Microsoft Corporation. 

3# Licensed under the MIT License. 

4# 

5""" 

6Factory method to create a new :py:class:`.Storage` instance from a 

7:py:attr:`~mlos_bench.config.schemas.config_schemas.ConfigSchema.STORAGE` type json 

8config. 

9 

10See Also 

11-------- 

12mlos_bench.storage : For example usage. 

13""" 

14 

15from typing import Any, Dict, List, Optional 

16 

17from mlos_bench.config.schemas import ConfigSchema 

18from mlos_bench.services.config_persistence import ConfigPersistenceService 

19from mlos_bench.storage.base_storage import Storage 

20 

21 

22def from_config( 

23 config: str, 

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

25 **kwargs: Any, 

26) -> Storage: 

27 """ 

28 Create a new storage object from JSON5 config file. 

29 

30 Parameters 

31 ---------- 

32 config : str 

33 JSON5 config file or string to load. 

34 global_configs : Optional[List[str]] 

35 An optional list of config files with global parameters. 

36 kwargs : dict 

37 Additional configuration parameters. 

38 

39 Returns 

40 ------- 

41 storage : Storage 

42 A new storage object. 

43 """ 

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

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

46 global_config = {} 

47 for fname in global_configs or []: 

48 gconfig = config_loader.load_config(fname, ConfigSchema.GLOBALS) 

49 global_config.update(gconfig) 

50 config_path += gconfig.get("config_path", []) 

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

52 global_config.update(kwargs) 

53 

54 class_config = config_loader.load_config(config, ConfigSchema.STORAGE) 

55 assert isinstance(class_config, Dict) 

56 

57 ret = config_loader.build_storage( 

58 service=config_loader, 

59 config=class_config, 

60 global_config=global_config, 

61 ) 

62 assert isinstance(ret, Storage) 

63 return ret