Coverage for mlos_bench/mlos_bench/storage/storage_factory.py: 26%
19 statements
« prev ^ index » next coverage.py v7.6.1, created at 2024-10-07 01:52 +0000
« prev ^ index » next coverage.py v7.6.1, created at 2024-10-07 01:52 +0000
1#
2# Copyright (c) Microsoft Corporation.
3# Licensed under the MIT License.
4#
5"""Factory method to create a new Storage instance from configs."""
7from typing import Any, Dict, List, Optional
9from mlos_bench.config.schemas import ConfigSchema
10from mlos_bench.services.config_persistence import ConfigPersistenceService
11from mlos_bench.storage.base_storage import Storage
14def from_config(
15 config_file: str,
16 global_configs: Optional[List[str]] = None,
17 **kwargs: Any,
18) -> Storage:
19 """
20 Create a new storage object from JSON5 config file.
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.
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)
46 class_config = config_loader.load_config(config_file, ConfigSchema.STORAGE)
47 assert isinstance(class_config, Dict)
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