Coverage for railway / core / settings.py: 88%

1"""Settings management for Railway Framework.""" 

2 

3from pathlib import Path 

4from typing import Any 

5 

6import yaml 

7from pydantic import BaseModel, Field 

8from pydantic_settings import BaseSettings, SettingsConfigDict 

9 

10 

11class APISettings(BaseModel): 

12 """API configuration.""" 

13 

14 base_url: str = "" 

15 timeout: int = 30 

16 max_retries: int = 3 

17 

18 

19class DatabaseSettings(BaseModel): 

20 """Database configuration.""" 

21 

22 host: str = "localhost" 

23 port: int = 5432 

24 name: str = "db" 

25 user: str | None = None 

26 password: str | None = None 

27 

28 

29class RetryNodeSettings(BaseModel): 

30 """Per-node retry settings.""" 

31 

32 max_attempts: int = 3 

33 min_wait: int = 2 

34 max_wait: int = 10 

35 multiplier: int = 1 

36 

37 

38class RetrySettings(BaseModel): 

39 """Retry policy configuration.""" 

40 

41 default: RetryNodeSettings = Field(default_factory=RetryNodeSettings) 

42 nodes: dict[str, RetryNodeSettings] = Field(default_factory=dict) 

43 

44 

45class LoggingHandlerSettings(BaseModel): 

46 """Logging handler configuration.""" 

47 

48 type: str # file, console 

49 level: str = "INFO" 

50 path: str | None = None 

51 rotation: str | None = None 

52 retention: str | None = None 

53 

54 

55class LoggingSettings(BaseModel): 

56 """Logging configuration.""" 

57 

58 level: str = "INFO" 

59 format: str = "{time:HH:mm:ss} | {level} | {message}" 

60 handlers: list[LoggingHandlerSettings] = Field(default_factory=list) 

61 

62 

63def _load_yaml_config(config_dir: Path, env: str) -> dict[str, Any]: 

64 """ 

65 Load YAML configuration file. 

66 

67 Args: 

68 config_dir: Directory containing config files 

69 env: Environment name (development, production, etc.) 

70 

71 Returns: 

72 Configuration dictionary 

73 """ 

74 config_file = config_dir / f"{env}.yaml" 

75 

76 if not config_file.exists(): 

77 return {} 

78 

79 with open(config_file, encoding="utf-8") as f: 

80 return yaml.safe_load(f) or {} 

81 

82 

83def _parse_api_settings(config: dict[str, Any]) -> APISettings: 

84 """Parse API settings from config dict.""" 

85 api_config = config.get("api", {}) 

86 return APISettings(**api_config) if api_config else APISettings() 

87 

88 

89def _parse_database_settings(config: dict[str, Any]) -> DatabaseSettings: 

90 """Parse database settings from config dict.""" 

91 db_config = config.get("database", {}) 

92 return DatabaseSettings(**db_config) if db_config else DatabaseSettings() 

93 

94 

95def _parse_retry_settings(config: dict[str, Any]) -> RetrySettings: 

96 """Parse retry settings from config dict.""" 

97 retry_config = config.get("retry", {}) 

98 if not retry_config: 

99 return RetrySettings() 

100 

101 default_retry = RetryNodeSettings(**retry_config.get("default", {})) 

102 nodes_retry = { 

103 name: RetryNodeSettings(**settings) 

104 for name, settings in retry_config.get("nodes", {}).items() 

105 } 

106 return RetrySettings(default=default_retry, nodes=nodes_retry) 

107 

108 

109def _parse_logging_settings(config: dict[str, Any]) -> LoggingSettings: 

110 """Parse logging settings from config dict.""" 

111 log_config = config.get("logging", {}) 

112 if not log_config: 

113 return LoggingSettings() 

114 

115 handlers = [ 

116 LoggingHandlerSettings(**h) for h in log_config.get("handlers", []) 

117 ] 

118 return LoggingSettings( 

119 level=log_config.get("level", "INFO"), 

120 format=log_config.get("format", "{time:HH:mm:ss} | {level} | {message}"), 

121 handlers=handlers, 

122 ) 

123 

124 

125def _apply_env_overrides( 

126 logging_settings: LoggingSettings, 

127 log_level: str | None, 

128) -> LoggingSettings: 

129 """ 

130 Apply environment variable overrides to logging settings. 

131 

132 Returns new LoggingSettings with overrides applied. 

133 """ 

134 if log_level: 

135 return LoggingSettings( 

136 level=log_level, 

137 format=logging_settings.format, 

138 handlers=logging_settings.handlers, 

139 ) 

140 return logging_settings 

141 

142 

143class Settings(BaseSettings): 

144 """ 

145 Application settings. 

146 

147 Loads configuration from: 

148 1. Environment variables (.env file) 

149 2. YAML config file (config/{env}.yaml) 

150 

151 Environment variables override YAML config values. 

152 """ 

153 

154 model_config = SettingsConfigDict( 

155 env_file=".env", 

156 env_file_encoding="utf-8", 

157 case_sensitive=False, 

158 extra="allow", 

159 ) 

160 

161 # Environment variables 

162 railway_env: str = "development" 

163 app_name: str = "railway_app" 

164 log_level: str | None = None # Override from .env 

165 

166 # Configuration from YAML (will be populated after init) 

167 api: APISettings = Field(default_factory=APISettings) 

168 database: DatabaseSettings = Field(default_factory=DatabaseSettings) 

169 retry: RetrySettings = Field(default_factory=RetrySettings) 

170 logging: LoggingSettings = Field(default_factory=LoggingSettings) 

171 

172 def __init__(self, _config_dir: str | None = None, **kwargs: Any) -> None: 

173 """Initialize settings and load YAML config.""" 

174 super().__init__(**kwargs) 

175 

176 # Determine config directory 

177 config_dir = self._resolve_config_dir(_config_dir) 

178 

179 # Load and parse YAML configuration 

180 config = _load_yaml_config(config_dir, self.railway_env) 

181 

182 # Parse individual sections (functional approach) 

183 self.api = _parse_api_settings(config) 

184 self.database = _parse_database_settings(config) 

185 self.retry = _parse_retry_settings(config) 

186 self.logging = _parse_logging_settings(config) 

187 

188 # Apply environment variable overrides 

189 self.logging = _apply_env_overrides(self.logging, self.log_level) 

190 

191 def _resolve_config_dir(self, config_dir: str | None) -> Path: 

192 """Resolve the configuration directory path.""" 

193 if config_dir: 193 ↛ 197line 193 didn't jump to line 197 because the condition on line 193 was always true

194 return Path(config_dir) 

195 

196 # Default: look for config/ in current directory or parent 

197 cwd_config = Path.cwd() / "config" 

198 if cwd_config.exists(): 

199 return cwd_config 

200 

201 return Path(__file__).parent.parent.parent / "config" 

202 

203 def get_retry_settings(self, node_name: str) -> RetryNodeSettings: 

204 """ 

205 Get retry settings for a specific node. 

206 

207 Args: 

208 node_name: Name of the node 

209 

210 Returns: 

211 RetryNodeSettings for the node, or default if not specified 

212 """ 

213 return self.retry.nodes.get(node_name, self.retry.default) 

214 

215 

216# Global settings instance (lazy initialization) 

217_settings: Settings | None = None 

218 

219 

220def get_settings(_config_dir: str | None = None) -> Settings: 

221 """ 

222 Get the global settings instance. 

223 

224 Creates a new instance on first call, returns cached instance thereafter. 

225 

226 Args: 

227 _config_dir: Optional config directory (only used on first call) 

228 

229 Returns: 

230 Settings instance 

231 """ 

232 global _settings 

233 if _settings is None: 

234 _settings = Settings(_config_dir=_config_dir) 

235 return _settings 

236 

237 

238def reset_settings() -> None: 

239 """Reset the global settings instance (for testing).""" 

240 global _settings 

241 _settings = None