Coverage for llm_dataset_engine/config/config_loader.py: 45%

1""" 

2Configuration loader for YAML and JSON files. 

3 

4Enables loading pipeline configurations from declarative files. 

5""" 

6 

7import json 

8from pathlib import Path 

9from typing import Any, Dict 

10 

11import yaml 

12 

13from llm_dataset_engine.api.pipeline import Pipeline 

14from llm_dataset_engine.core.specifications import PipelineSpecifications 

15 

16 

17class ConfigLoader: 

18 """ 

19 Loads pipeline configurations from YAML or JSON files. 

20  

21 Follows Single Responsibility: only handles config file loading. 

22 """ 

23 

24 @staticmethod 

25 def from_yaml(file_path: str | Path) -> PipelineSpecifications: 

26 """ 

27 Load configuration from YAML file. 

28 

29 Args: 

30 file_path: Path to YAML file 

31 

32 Returns: 

33 PipelineSpecifications 

34 

35 Raises: 

36 FileNotFoundError: If file doesn't exist 

37 ValueError: If invalid YAML or configuration 

38 """ 

39 path = Path(file_path) 

40 

41 if not path.exists(): 

42 raise FileNotFoundError(f"Config file not found: {path}") 

43 

44 with open(path, "r") as f: 

45 config_dict = yaml.safe_load(f) 

46 

47 return ConfigLoader._dict_to_specifications(config_dict) 

48 

49 @staticmethod 

50 def from_json(file_path: str | Path) -> PipelineSpecifications: 

51 """ 

52 Load configuration from JSON file. 

53 

54 Args: 

55 file_path: Path to JSON file 

56 

57 Returns: 

58 PipelineSpecifications 

59 

60 Raises: 

61 FileNotFoundError: If file doesn't exist 

62 ValueError: If invalid JSON or configuration 

63 """ 

64 path = Path(file_path) 

65 

66 if not path.exists(): 

67 raise FileNotFoundError(f"Config file not found: {path}") 

68 

69 with open(path, "r") as f: 

70 config_dict = json.load(f) 

71 

72 return ConfigLoader._dict_to_specifications(config_dict) 

73 

74 @staticmethod 

75 def _dict_to_specifications( 

76 config: Dict[str, Any] 

77 ) -> PipelineSpecifications: 

78 """ 

79 Convert configuration dictionary to PipelineSpecifications. 

80 

81 Args: 

82 config: Configuration dictionary 

83 

84 Returns: 

85 PipelineSpecifications 

86 """ 

87 return PipelineSpecifications(**config) 

88 

89 @staticmethod 

90 def to_yaml( 

91 specifications: PipelineSpecifications, file_path: str | Path 

92 ) -> None: 

93 """ 

94 Save specifications to YAML file. 

95 

96 Args: 

97 specifications: Pipeline specifications 

98 file_path: Destination file path 

99 """ 

100 path = Path(file_path) 

101 

102 # Convert to dict 

103 config_dict = specifications.model_dump(mode="json") 

104 

105 with open(path, "w") as f: 

106 yaml.dump(config_dict, f, default_flow_style=False, indent=2) 

107 

108 @staticmethod 

109 def to_json( 

110 specifications: PipelineSpecifications, file_path: str | Path 

111 ) -> None: 

112 """ 

113 Save specifications to JSON file. 

114 

115 Args: 

116 specifications: Pipeline specifications 

117 file_path: Destination file path 

118 """ 

119 path = Path(file_path) 

120 

121 # Convert to dict 

122 config_dict = specifications.model_dump(mode="json") 

123 

124 with open(path, "w") as f: 

125 json.dump(config_dict, f, indent=2, default=str) 

126