Coverage for packages/core/src/langgate/core/utils/config_utils.py: 57%

1"""Configuration utilities for LangGate.""" 

2 

3import os 

4from pathlib import Path 

5from typing import TypeVar 

6 

7import yaml 

8from pydantic import BaseModel, ValidationError 

9 

10from langgate.core.logging import StructLogger 

11 

12ConfigSchemaT = TypeVar("ConfigSchemaT", bound=BaseModel) 

13 

14 

15def resolve_path( 

16 env_var: str, 

17 arg_path: Path | None = None, 

18 default_path: Path | None = None, 

19 path_desc: str = "path", 

20 logger: StructLogger | None = None, 

21) -> Path: 

22 """Resolve a file path based on priority: args > env > default. 

23 

24 Args: 

25 env_var: Environment variable name to check 

26 arg_path: Path provided in constructor args 

27 default_path: Default path to use if others not provided 

28 path_desc: Description for logging 

29 logger: Optional logger instance for recording path resolution 

30 

31 Returns: 

32 Resolved Path object 

33 """ 

34 # Priority: args > env > default 

35 resolved_path = arg_path or Path(os.getenv(env_var, str(default_path))) 

36 

37 # Log the resolved path and its existence 

38 if logger: 

39 exists = resolved_path.exists() 

40 logger.debug( 

41 f"resolved_{path_desc}", 

42 path=str(resolved_path), 

43 exists=exists, 

44 source="args" if arg_path else ("env" if os.getenv(env_var) else "default"), 

45 ) 

46 

47 return resolved_path 

48 

49 

50def load_yaml_config( 

51 config_path: Path, 

52 schema_class: type[ConfigSchemaT], 

53 logger: StructLogger | None = None, 

54) -> ConfigSchemaT: 

55 """Load and validate a YAML configuration file using a Pydantic schema. 

56 

57 Args: 

58 config_path: Path to the YAML configuration file 

59 schema_class: The Pydantic schema class to validate against 

60 logger: Optional logger instance for recording validation results 

61 

62 Returns: 

63 The validated schema instance 

64 

65 Raises: 

66 FileNotFoundError: If config file doesn't exist 

67 ValueError: If config file is empty 

68 yaml.YAMLError: If YAML parsing fails 

69 ValidationError: If schema validation fails 

70 """ 

71 try: 

72 if not config_path.exists(): 

73 if logger: 

74 logger.error( 

75 "config_file_not_found", 

76 config_path=str(config_path), 

77 ) 

78 raise FileNotFoundError(f"Config file not found: {config_path}") 

79 

80 with open(config_path) as f: 

81 raw_config = yaml.safe_load(f) 

82 

83 if raw_config is None: 

84 if logger: 

85 logger.error("config_file_is_empty", config_path=str(config_path)) 

86 raise ValueError(f"Config file is empty: {config_path}") 

87 

88 # Validate configuration using Pydantic schema 

89 try: 

90 config = schema_class.model_validate(raw_config) 

91 if logger: 

92 logger.info( 

93 "loaded_config", 

94 config_path=str(config_path), 

95 ) 

96 return config 

97 except ValidationError as exc: 

98 if logger: 

99 logger.exception( 

100 "invalid_config_format", 

101 config_path=str(config_path), 

102 errors=str(exc), 

103 ) 

104 raise 

105 

106 except yaml.YAMLError: 

107 if logger: 

108 logger.exception( 

109 "failed_to_parse_yaml_config", config_path=str(config_path) 

110 ) 

111 raise 

112 except Exception: 

113 if logger: 

114 logger.exception("failed_to_load_config", config_path=str(config_path)) 

115 raise