Coverage for src / local_deep_research / web / server_config.py: 74%

1"""Server configuration management for web app startup. 

2 

3This module handles server configuration that needs to be available before 

4Flask app context is established. It provides a bridge between environment 

5variables and database settings. 

6""" 

7 

8import json 

9from pathlib import Path 

10from typing import Dict, Any 

11 

12from loguru import logger 

13 

14from ..config.paths import get_data_dir 

15from ..settings.manager import get_typed_setting_value 

16 

17 

18def get_server_config_path() -> Path: 

19 """Get the path to the server configuration file.""" 

20 return Path(get_data_dir()) / "server_config.json" 

21 

22 

23def load_server_config() -> Dict[str, Any]: 

24 """Load server configuration from file or environment variables. 

25 

26 Returns: 

27 dict: Server configuration with keys: host, port, debug, use_https 

28 """ 

29 config_path = get_server_config_path() 

30 

31 # Try to load from config file first 

32 saved_config = {} 

33 if config_path.exists(): 33 ↛ 34line 33 didn't jump to line 34 because the condition on line 33 was never true

34 try: 

35 with open(config_path, "r") as f: 

36 saved_config = json.load(f) 

37 logger.debug(f"Loaded server config from {config_path}") 

38 except Exception as e: 

39 logger.warning(f"Failed to load server config: {e}") 

40 

41 # Ensure correct typing and check environment variables. 

42 config = { 

43 "host": get_typed_setting_value( 

44 "web.host", saved_config.get("host"), "text", default="0.0.0.0" 

45 ), 

46 "port": get_typed_setting_value( 

47 "web.port", saved_config.get("port"), "number", default=5000 

48 ), 

49 "debug": get_typed_setting_value( 

50 "app.debug", saved_config.get("debug"), "checkbox", default=False 

51 ), 

52 "use_https": get_typed_setting_value( 

53 "web.use_https", 

54 saved_config.get("use_https"), 

55 "checkbox", 

56 default=True, 

57 ), 

58 "allow_registrations": get_typed_setting_value( 

59 "app.allow_registrations", 

60 saved_config.get("allow_registrations"), 

61 "checkbox", 

62 default=True, 

63 ), 

64 # Rate limiting settings 

65 "rate_limit_default": get_typed_setting_value( 

66 "security.rate_limit_default", 

67 saved_config.get("rate_limit_default"), 

68 "text", 

69 default="5000 per hour;50000 per day", 

70 ), 

71 "rate_limit_login": get_typed_setting_value( 

72 "security.rate_limit_login", 

73 saved_config.get("rate_limit_login"), 

74 "text", 

75 default="5 per 15 minutes", 

76 ), 

77 "rate_limit_registration": get_typed_setting_value( 

78 "security.rate_limit_registration", 

79 saved_config.get("rate_limit_registration"), 

80 "text", 

81 default="3 per hour", 

82 ), 

83 } 

84 

85 return config 

86 

87 

88def save_server_config(config: Dict[str, Any]) -> None: 

89 """Save server configuration to file. 

90 

91 This should be called when web.host or web.port settings are updated 

92 through the UI. 

93 

94 Args: 

95 config: Server configuration dict 

96 """ 

97 config_path = get_server_config_path() 

98 

99 try: 

100 # Ensure directory exists 

101 config_path.parent.mkdir(parents=True, exist_ok=True) 

102 

103 with open(config_path, "w") as f: 

104 json.dump(config, f, indent=2) 

105 

106 logger.info(f"Saved server config to {config_path}") 

107 except Exception: 

108 logger.exception("Failed to save server config") 

109 

110 

111def sync_from_settings(settings_snapshot: Dict[str, Any]) -> None: 

112 """Sync server config from settings snapshot. 

113 

114 This should be called when settings are updated through the UI. 

115 

116 Args: 

117 settings_snapshot: Settings snapshot containing web.host and web.port 

118 """ 

119 config = load_server_config() 

120 

121 # Update from settings if available 

122 if "web.host" in settings_snapshot: 122 ↛ 124line 122 didn't jump to line 124 because the condition on line 122 was always true

123 config["host"] = settings_snapshot["web.host"] 

124 if "web.port" in settings_snapshot: 124 ↛ 126line 124 didn't jump to line 126 because the condition on line 124 was always true

125 config["port"] = settings_snapshot["web.port"] 

126 if "app.debug" in settings_snapshot: 126 ↛ 128line 126 didn't jump to line 128 because the condition on line 126 was always true

127 config["debug"] = settings_snapshot["app.debug"] 

128 if "web.use_https" in settings_snapshot: 128 ↛ 130line 128 didn't jump to line 130 because the condition on line 128 was always true

129 config["use_https"] = settings_snapshot["web.use_https"] 

130 if "app.allow_registrations" in settings_snapshot: 130 ↛ 134line 130 didn't jump to line 134 because the condition on line 130 was always true

131 config["allow_registrations"] = settings_snapshot[ 

132 "app.allow_registrations" 

133 ] 

134 if "security.rate_limit_default" in settings_snapshot: 134 ↛ 138line 134 didn't jump to line 138 because the condition on line 134 was always true

135 config["rate_limit_default"] = settings_snapshot[ 

136 "security.rate_limit_default" 

137 ] 

138 if "security.rate_limit_login" in settings_snapshot: 138 ↛ 142line 138 didn't jump to line 142 because the condition on line 138 was always true

139 config["rate_limit_login"] = settings_snapshot[ 

140 "security.rate_limit_login" 

141 ] 

142 if "security.rate_limit_registration" in settings_snapshot: 142 ↛ 147line 142 didn't jump to line 147 because the condition on line 142 was always true

143 config["rate_limit_registration"] = settings_snapshot[ 

144 "security.rate_limit_registration" 

145 ] 

146 

147 save_server_config(config)