Coverage for railway / core / config.py: 90%

1""" 

2Settings provider registry for framework-user code separation. 

3 

4This module allows the @node decorator to access user settings 

5without directly importing user code. 

6 

7Features: 

8- Lazy initialization via _SettingsProxy 

9- Thread-safe settings access 

10- Cached settings for performance 

11""" 

12 

13import threading 

14from collections.abc import Callable 

15from pathlib import Path 

16from types import SimpleNamespace 

17from typing import Any, Protocol, cast 

18 

19 

20class RetrySettingsProtocol(Protocol): 

21 """Protocol for retry settings objects.""" 

22 

23 max_attempts: int 

24 min_wait: int 

25 max_wait: int 

26 multiplier: int 

27 

28 

29# Thread-safe settings state 

30_settings_provider: Callable[[], Any] | None = None 

31_settings_cache: Any | None = None 

32_settings_lock = threading.Lock() 

33 

34 

35def register_settings_provider(provider: Callable[[], Any]) -> None: 

36 """ 

37 Register a settings provider function. 

38 

39 The provider should return a settings object with get_retry_settings() method. 

40 

41 Args: 

42 provider: A callable that returns the settings object 

43 

44 Example: 

45 from railway.core.config import register_settings_provider 

46 from src.settings import get_settings 

47 

48 register_settings_provider(get_settings) 

49 """ 

50 global _settings_provider 

51 _settings_provider = provider 

52 

53 

54def get_settings_provider() -> Callable[[], Any] | None: 

55 """ 

56 Get the registered settings provider. 

57 

58 Returns: 

59 The registered provider function, or None if not registered. 

60 """ 

61 return _settings_provider 

62 

63 

64class DefaultRetrySettings: 

65 """ 

66 Default retry settings when no provider is registered. 

67 

68 Used as fallback when: 

69 - No settings provider is registered 

70 - The provider raises an exception 

71 """ 

72 

73 max_attempts: int = 3 

74 min_wait: int = 2 

75 max_wait: int = 10 

76 multiplier: int = 1 

77 

78 

79def get_retry_config(node_name: str) -> RetrySettingsProtocol: 

80 """ 

81 Get retry configuration for a specific node. 

82 

83 If no settings provider is registered, returns default settings. 

84 

85 Args: 

86 node_name: Name of the node to get settings for 

87 

88 Returns: 

89 Retry configuration object with max_attempts, min_wait, max_wait, multiplier 

90 """ 

91 if _settings_provider is None: 

92 return DefaultRetrySettings() 

93 

94 try: 

95 settings = _settings_provider() 

96 return cast(RetrySettingsProtocol, settings.get_retry_settings(node_name)) 

97 except Exception: 

98 return DefaultRetrySettings() 

99 

100 

101def reset_provider() -> None: 

102 """ 

103 Reset the settings provider (for testing). 

104 """ 

105 global _settings_provider 

106 _settings_provider = None 

107 

108 

109def reset_settings() -> None: 

110 """ 

111 Reset settings cache and provider. 

112 

113 Forces settings to be reloaded on next access. 

114 Useful for testing or when environment changes. 

115 """ 

116 global _settings_cache, _settings_provider 

117 with _settings_lock: 

118 _settings_cache = None 

119 _settings_provider = None 

120 

121 

122def _get_or_create_settings() -> Any: 

123 """ 

124 Get or create the settings object. 

125 

126 Uses registered provider if available, otherwise returns default settings. 

127 Thread-safe and cached for performance. 

128 

129 Returns: 

130 The settings object. 

131 """ 

132 global _settings_cache 

133 

134 with _settings_lock: 

135 if _settings_cache is not None: 

136 return _settings_cache 

137 

138 # Use registered provider if available 

139 if _settings_provider is not None: 139 ↛ 147line 139 didn't jump to line 147 because the condition on line 139 was always true

140 try: 

141 _settings_cache = _settings_provider() 

142 return _settings_cache 

143 except Exception: 

144 pass 

145 

146 # Return default settings 

147 _settings_cache = _create_default_settings() 

148 return _settings_cache 

149 

150 

151def _create_default_settings() -> Any: 

152 """Create default settings object.""" 

153 return SimpleNamespace( 

154 api=SimpleNamespace( 

155 base_url="", 

156 timeout=30, 

157 ), 

158 retry=SimpleNamespace( 

159 default=SimpleNamespace( 

160 max_attempts=3, 

161 min_wait=2.0, 

162 max_wait=10.0, 

163 multiplier=2, 

164 ), 

165 nodes={}, 

166 ), 

167 logging=SimpleNamespace( 

168 level="INFO", 

169 format="console", 

170 ), 

171 ) 

172 

173 

174class _SettingsProxy: 

175 """ 

176 Proxy object for lazy settings initialization. 

177 

178 Settings are not loaded until first attribute access. 

179 This avoids import-time side effects and circular imports. 

180 

181 Usage: 

182 from railway.core.config import settings 

183 

184 # Settings loaded here (first access) 

185 api_url = settings.api.base_url 

186 """ 

187 

188 def __getattr__(self, name: str) -> Any: 

189 """Delegate attribute access to actual settings object.""" 

190 actual_settings = _get_or_create_settings() 

191 return getattr(actual_settings, name) 

192 

193 def __repr__(self) -> str: 

194 return "<_SettingsProxy>" 

195 

196 

197# Module-level settings proxy (not initialized until first access) 

198settings = _SettingsProxy()