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 | None: 

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 or None if validation failed 

64 """ 

65 try: 

66 if not config_path.exists(): 

67 if logger: 

68 logger.warning( 

69 "config_file_not_found", 

70 config_path=str(config_path), 

71 ) 

72 return None 

73 

74 with open(config_path) as f: 

75 raw_config = yaml.safe_load(f) 

76 

77 if raw_config is None: 

78 if logger: 

79 logger.warning("config_file_is_empty", config_path=str(config_path)) 

80 return None 

81 

82 # Validate configuration using Pydantic schema 

83 try: 

84 config = schema_class.model_validate(raw_config) 

85 if logger: 

86 logger.info( 

87 "loaded_config", 

88 config_path=str(config_path), 

89 ) 

90 return config 

91 except ValidationError as exc: 

92 if logger: 

93 logger.exception( 

94 "invalid_config_format", 

95 config_path=str(config_path), 

96 errors=str(exc), 

97 ) 

98 return None 

99 

100 except yaml.YAMLError: 

101 if logger: 

102 logger.exception( 

103 "failed_to_parse_yaml_config", config_path=str(config_path) 

104 ) 

105 return None 

106 except Exception: 

107 if logger: 

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

109 return None