Coverage for src/lumen/config.py: 100%

1""" 

2Using pydantic BaseSettings to have custom settings for the app. 

3BaseSettings has the ability to read from environment variables and .env files. 

4""" 

5 

6import os 

7import sys 

8from pathlib import Path 

9from uuid import UUID 

10 

11from pydantic import AliasChoices, Field 

12from pydantic_settings import BaseSettings, SettingsConfigDict 

13 

14SECRETS_DIR = "/run/secrets" if Path("/run/secrets").exists() else None 

15 

16# Do not load .env file when running tests to ensure isolation and reproducibility 

17IS_TESTING = "pytest" in sys.modules or "PYTEST_CURRENT_TEST" in os.environ 

18ENV_FILE = None if IS_TESTING else ".env" 

19 

20 

21class DatabaseSettings(BaseSettings): 

22 """ 

23 Main database settings 

24 """ 

25 

26 model_config = SettingsConfigDict( 

27 env_prefix="LUMEN_DB_", 

28 env_file=ENV_FILE, 

29 env_file_encoding="utf-8", 

30 # support docker secrets (currently not used) 

31 secrets_dir=SECRETS_DIR, 

32 populate_by_name=True, 

33 extra="ignore", 

34 ) 

35 url: str = Field( 

36 default="postgresql://lumen:lumen@localhost:5432/lumen", 

37 validation_alias=AliasChoices("lumen_db_url", "LUMEN_DB_URL"), 

38 ) 

39 min_connections: int = Field(default=2, ge=1) 

40 max_connections: int = Field(default=10, ge=1) 

41 

42 

43class SmartSettings(BaseSettings): 

44 """SMART-on-FHIR resource-server configuration (Phase 12).""" 

45 

46 model_config = SettingsConfigDict( 

47 env_prefix="LUMEN_SMART_", 

48 env_file=ENV_FILE, 

49 env_file_encoding="utf-8", 

50 extra="ignore", 

51 ) 

52 

53 enabled: bool = False 

54 audience: str = "lumen" 

55 jwks_cache_ttl: int = Field(default=300, ge=10) 

56 

57 # Maps OIDC issuer URL → tenant UUID. 

58 # Example (env): LUMEN_SMART_TENANT_MAP='{"https://kc.example.com/realms/a":"<uuid>"}' 

59 tenant_map: dict[str, UUID] = Field(default_factory=dict) 

60 

61 # Maps external issuer URL → internal base URL for JWKS/OIDC discovery fetching. 

62 # Use when Lumen runs in Docker and cannot reach the external issuer URL directly. 

63 # Example: LUMEN_SMART_INTERNAL_ISSUER_MAP='{"http://localhost:8180/realms/r":"http://keycloak:8080/realms/r"}' 

64 internal_issuer_map: dict[str, str] = Field(default_factory=dict) 

65 

66 

67class ObservabilitySettings(BaseSettings): 

68 """OpenTelemetry / OTLP exporter configuration (Phase 13b). Disabled by default.""" 

69 

70 model_config = SettingsConfigDict( 

71 env_prefix="LUMEN_OTEL_", 

72 env_file=ENV_FILE, 

73 env_file_encoding="utf-8", 

74 extra="ignore", 

75 ) 

76 

77 enabled: bool = False 

78 # OTLP/HTTP endpoint base (e.g. http://otel-lgtm:4318). No trailing slash. 

79 exporter_endpoint: str = "http://localhost:4318" 

80 service_name: str = "lumen" 

81 traces_sample_rate: float = Field(default=1.0, ge=0.0, le=1.0) 

82 # Optional OTLP log export (Weg A keeps structlog-JSON on stdout by default). 

83 log_export: bool = False 

84 

85 

86class TenantEditableSettings(BaseSettings): 

87 """Settings that can be overridden at the tenant level and modified via the API.""" 

88 

89 model_config = SettingsConfigDict( 

90 env_prefix="LUMEN_", 

91 env_file=ENV_FILE, 

92 env_file_encoding="utf-8", 

93 extra="ignore", 

94 ) 

95 

96 create_on_update: bool = True 

97 # Recommended baseline (docs: knowledge/VALIDIERUNG.md): 

98 # keep the two cheap, "pure FHIR conformance" checks strict (structure via 

99 # JSON Schema, and base R4B code enums — both free, no DB I/O), while the 

100 # profile-/reference-specific checks stay opt-in for performance. 

101 reference_integrity: str = "disabled" # "strict" | "lenient" | "disabled" 

102 validation_strictness: str = "strict" # "strict" | "lenient" — JSON Schema structure 

103 terminology_strictness: str = "strict" # "strict" | "lenient" — base R4B code bindings 

104 timezone_strictness: str = "disabled" # "strict" | "lenient" | "disabled" 

105 profile_validation_mode: str = "tolerant" # "tolerant" | "declared" | "enforced" 

106 enforced_profiles: dict[str, list[str]] = Field(default_factory=dict) 

107 ignore_unknown_search_parameters: bool = False 

108 # Stage 3b (Terminology Bindings, docs/architecture/profile-validation.md): opt-in 

109 # per tenant, disabled by default — Lumen stays 100% offline unless a tenant 

110 # explicitly allows outbound $expand calls to a public terminology server for 

111 # ValueSets the local engine can't resolve. 

112 terminology_pre_expansion_enabled: bool = False 

113 

114 

115class TerminologySettings(BaseSettings): 

116 """External terminology server config for Stage 3b Pre-Expansion. Not 

117 tenant-editable (a deployment-wide server choice, unlike the per-tenant 

118 opt-in flag `TenantEditableSettings.terminology_pre_expansion_enabled`).""" 

119 

120 model_config = SettingsConfigDict( 

121 env_prefix="LUMEN_TERMINOLOGY_", 

122 env_file=ENV_FILE, 

123 env_file_encoding="utf-8", 

124 extra="ignore", 

125 ) 

126 

127 server_url: str = "https://tx.fhir.org/r4" 

128 timeout_seconds: float = Field(default=10.0, gt=0.0) 

129 

130 

131class RateLimitSettings(BaseSettings): 

132 """In-memory, per-process rate limiting + request-size limiting (Phase 16 

133 hardening). Not tenant-editable — a deployment-wide operational safeguard. 

134 Enabled by default with a generous limit; per-process only (no Redis) — 

135 each instance in a horizontally-scaled deployment counts independently. 

136 """ 

137 

138 model_config = SettingsConfigDict( 

139 env_prefix="LUMEN_RATELIMIT_", 

140 env_file=ENV_FILE, 

141 env_file_encoding="utf-8", 

142 extra="ignore", 

143 ) 

144 

145 enabled: bool = True 

146 requests_per_minute: int = Field(default=300, ge=1) 

147 max_body_bytes: int = Field(default=50 * 1024 * 1024, ge=1) # 50 MB 

148 

149 

150class SecurityHeadersSettings(BaseSettings): 

151 """Baseline browser security headers (Phase 16 hardening). Not tenant-editable.""" 

152 

153 model_config = SettingsConfigDict( 

154 env_prefix="LUMEN_SECURITY_", 

155 env_file=ENV_FILE, 

156 env_file_encoding="utf-8", 

157 extra="ignore", 

158 ) 

159 

160 enabled: bool = True 

161 hsts_max_age_seconds: int = Field(default=63072000, ge=0) # 2 years 

162 

163 

164class CortexSettings(BaseSettings): 

165 """In-process plugin host (Phase 18). Disabled by default — a server with no 

166 plugins configured behaves exactly as it did before Cortex existed.""" 

167 

168 model_config = SettingsConfigDict( 

169 env_prefix="LUMEN_CORTEX_", 

170 env_file=ENV_FILE, 

171 env_file_encoding="utf-8", 

172 extra="ignore", 

173 ) 

174 

175 enabled: bool = False 

176 # Comma-separated "module:ClassName" (or bare "module") entries, e.g. 

177 # LUMEN_CORTEX_PLUGINS='lumen_plugins.sepsis_guard:SepsisGuard' 

178 plugins: str = "" 

179 # Also pick up plugins published under the "lumen.cortex.plugins" entry-point group. 

180 load_entry_points: bool = True 

181 # How deep hook-triggered writes may cascade. 1 means a plugin's own writes 

182 # do not fire hooks again — the safe default against accidental recursion. 

183 max_hook_depth: int = Field(default=1, ge=0, le=5) 

184 # Allow plugins to be registered at runtime (notebook DX). Development only: 

185 # it lets anything holding a reference to the process add server-side code. 

186 allow_runtime_registration: bool = False 

187 

188 

189class Settings(BaseSettings): 

190 """ 

191 Main app settings 

192 """ 

193 

194 model_config = SettingsConfigDict( 

195 env_prefix="LUMEN_", 

196 env_file=ENV_FILE, 

197 env_file_encoding="utf-8", 

198 # support docker secrets (currently not used) 

199 secrets_dir=SECRETS_DIR, 

200 extra="ignore", 

201 ) 

202 

203 server_address: str = "http://localhost:8000" 

204 fhir_version: str = "4.3.0" 

205 log_level: str = "info" 

206 log_format: str = "console" # "console" | "json" 

207 cors_origins: list[str] = Field(default=["*"]) 

208 seed_packages_dir: str | None = None 

209 seed_registry_packages: str | None = None 

210 

211 # Tenant used when no auth token supplies one (anonymous / single-tenant mode). 

212 default_tenant: UUID = UUID("00000000-0000-0000-0000-000000000001") 

213 

214 # tenant-specific settings (both API-editable and env-configurable) 

215 tenant: TenantEditableSettings = Field(default_factory=TenantEditableSettings) 

216 

217 # integrate settings for the database and SMART-on-FHIR resource-server 

218 db: DatabaseSettings = Field(default_factory=DatabaseSettings) 

219 smart: SmartSettings = Field(default_factory=SmartSettings) 

220 otel: ObservabilitySettings = Field(default_factory=ObservabilitySettings) 

221 terminology: TerminologySettings = Field(default_factory=TerminologySettings) 

222 ratelimit: RateLimitSettings = Field(default_factory=RateLimitSettings) 

223 security: SecurityHeadersSettings = Field(default_factory=SecurityHeadersSettings) 

224 cortex: CortexSettings = Field(default_factory=CortexSettings) 

225 

226 

227_settings: Settings | None = None 

228 

229 

230def get_settings() -> Settings: 

231 """ 

232 Get the settings for the app. 

233 If the settings are not set, set them. 

234 """ 

235 global _settings 

236 if _settings is None: 

237 _settings = Settings() 

238 return _settings