Coverage for packages/server/src/langgate/server/core/config.py: 79%

1import importlib.resources 

2import os 

3import secrets 

4from pathlib import Path 

5from typing import Any, Literal 

6 

7import yaml 

8from pydantic import SecretStr, field_validator 

9from pydantic.fields import FieldInfo 

10from pydantic_settings import ( 

11 BaseSettings, 

12 PydanticBaseSettingsSource, 

13 SettingsConfigDict, 

14) 

15 

16from langgate.core.fields import HttpUrlStr 

17from langgate.core.logging import get_logger 

18from langgate.core.utils.config_utils import resolve_path 

19 

20logger = get_logger(__name__) 

21 

22 

23class FixedYamlConfigSettingsSource(PydanticBaseSettingsSource): 

24 """Settings source that loads from app_config namespace in langgate_config.yaml in project root.""" 

25 

26 def __init__(self, settings_cls: type[BaseSettings]): 

27 super().__init__(settings_cls) 

28 cwd = Path.cwd() 

29 # Config path: env > cwd > package_dir 

30 core_resources = importlib.resources.files("langgate.core") 

31 default_config_path = Path( 

32 str(core_resources.joinpath("data", "default_config.yaml")) 

33 ) 

34 cwd_config_path = cwd / "langgate_config.yaml" 

35 

36 self.config_path = resolve_path( 

37 "LANGGATE_CONFIG", 

38 None, 

39 cwd_config_path if cwd_config_path.exists() else default_config_path, 

40 "server_config_path", 

41 logger, 

42 ) 

43 

44 def get_field_value( 

45 self, field: FieldInfo, field_name: str 

46 ) -> tuple[Any, str, bool]: 

47 """Get field value from app_config namespace in langgate_config.yaml file. 

48 

49 This is required by the base class but not used in our implementation. 

50 We use __call__ instead to load all settings at once. 

51 """ 

52 raise NotImplementedError 

53 

54 def prepare_field_value( 

55 self, field_name: str, field: FieldInfo, value: Any, value_is_complex: bool 

56 ) -> Any: 

57 return value 

58 

59 def __call__(self) -> dict[str, Any]: 

60 """Load settings from app_config namespace in langgate_config.yaml file.""" 

61 if not self.config_path.exists(): 

62 logger.warning(event="config_file_not_found", path=str(self.config_path)) 

63 return {} 

64 

65 try: 

66 with open(self.config_path) as f: 

67 config_data = yaml.safe_load(f) 

68 # Only return the app_config namespace 

69 app_config = config_data.get("app_config", {}) 

70 if not app_config: 

71 logger.warning( 

72 event="no_app_config_in_config_file", 

73 path=str(self.config_path), 

74 help="Add app_config namespace to yaml config file to override default settings", 

75 ) 

76 logger.info("loaded_app_config", settings=app_config) 

77 return app_config 

78 except Exception: 

79 logger.exception(event="app_config_load_error", path=str(self.config_path)) 

80 raise 

81 

82 

83def _get_api_env_file_path() -> Path | None: 

84 cwd = Path.cwd() 

85 cwd_env_path = cwd / ".env" 

86 env_file_path = resolve_path( 

87 "LANGGATE_ENV_FILE", None, cwd_env_path, "server_env_file", logger 

88 ) 

89 return env_file_path if env_file_path.exists() else None 

90 

91 

92class ApiSettings(BaseSettings): 

93 def __init__(self, *args, **kwargs): 

94 super().__init__(*args, **kwargs) 

95 

96 PROJECT_NAME: str = "LangGate" 

97 API_V1_STR: str = "/api/v1" 

98 LOG_LEVEL: Literal["debug", "info", "warning", "error", "critical"] = "info" 

99 # Ensure this is set and fixed across instances in production 

100 SECRET_KEY: SecretStr = SecretStr(secrets.token_urlsafe(32)) 

101 ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 7 

102 HTTPS: bool 

103 CORS_ORIGINS: list[HttpUrlStr | Literal["*"]] 

104 TEST_SERVER_HOST: str = "test" 

105 JSON_LOGS: bool = False 

106 IS_TEST: bool = False 

107 

108 @field_validator("LOG_LEVEL", mode="before") 

109 @classmethod 

110 def validate_log_level(cls, v: str | list[str]) -> list[str] | str: 

111 if isinstance(v, str): 

112 return v.lower() 

113 raise ValueError(f"Invalid log level {v}") 

114 

115 @field_validator("CORS_ORIGINS", mode="before") 

116 @classmethod 

117 def assemble_cors_origins(cls, v: str | list[str]) -> list[str] | str: 

118 if isinstance(v, str) and not v.startswith("["): 

119 return [i.strip() for i in v.split(",")] 

120 if isinstance(v, list | str): 

121 return v 

122 raise ValueError(v) 

123 

124 model_config = SettingsConfigDict( 

125 case_sensitive=True, 

126 env_file=_get_api_env_file_path(), 

127 env_file_encoding="utf-8", 

128 extra="ignore", 

129 ) 

130 

131 @classmethod 

132 def settings_customise_sources( 

133 cls, 

134 settings_cls: type[BaseSettings], 

135 init_settings: PydanticBaseSettingsSource, 

136 env_settings: PydanticBaseSettingsSource, 

137 dotenv_settings: PydanticBaseSettingsSource, 

138 file_secret_settings: PydanticBaseSettingsSource, 

139 ): 

140 return ( 

141 init_settings, # Highest priority: constructor arguments 

142 env_settings, # Second priority: environment variables 

143 dotenv_settings, # Third priority: .env file 

144 file_secret_settings, # Fourth priority: secrets from files 

145 # Lowest priority: YAML config file 

146 FixedYamlConfigSettingsSource(settings_cls), 

147 ) 

148 

149 def get_namespace(self, prefix: str, remove_prefix: bool = False) -> dict[str, Any]: 

150 namespace = {} 

151 for key, value in dict(self).items(): 

152 if key.startswith(prefix): 

153 if remove_prefix: 

154 namespace[key[len(prefix) :]] = value 

155 else: 

156 namespace[key] = value 

157 return namespace 

158 

159 

160class TestSettings(ApiSettings): 

161 def __init__(self, *args, **kwargs): 

162 logger.info("using_test_settings") 

163 BaseSettings.__init__(self, *args, **kwargs) 

164 

165 

166settings = ApiSettings() if not os.getenv("IS_TEST") else TestSettings()