Coverage for src / lexigram / admin / settings / service.py: 0%
227 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-11 02:25 +0800
1from __future__ import annotations
3from dataclasses import dataclass, field
4from datetime import UTC, datetime
5from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
7from lexigram import serialization as json
8from lexigram.admin.models.setting import SystemSetting
9from lexigram.admin.settings.panel.utils import map_config_node_type
10from lexigram.contracts.infra.cache import CacheBackendProtocol
11from lexigram.domain import DomainModel
12from lexigram.logging import get_logger
14logger = get_logger(__name__)
15_registry_warned = [False]
18def _warn_registry_unavailable() -> None:
19 """Emit a one-off warning when the admin panel registry is unavailable."""
21 if _registry_warned[0]:
22 return
23 _registry_warned[0] = True
24 logger.warning(
25 "config_registry_unavailable",
26 hint="pip install lexigram-admin",
27 detail="falling back to legacy settings resolution",
28 )
31@runtime_checkable
32class _SettingsRepository(Protocol):
33 """Narrow structural protocol for the settings storage backend.
35 Avoids a direct import of ``GenericRepository`` from ``lexigram-sql``.
36 """
38 async def find_one(self, **filters: Any) -> Any: ...
40 async def update(self, entity: Any) -> Any: ...
42 async def create(self, data: Any) -> Any: ...
45# Events integration
46from lexigram.contracts.domain import DomainEvent
47from lexigram.di.decorators import inject
49if TYPE_CHECKING:
50 from lexigram.contracts.core.di import ContainerResolverProtocol
53class SettingUpdated(DomainEvent):
54 """Fired when a setting is changed."""
56 key: str
57 value: str
58 scope: str
59 scope_id: str
62@dataclass(init=False)
63class SettingDefinition(DomainModel):
64 """Defines a comprehensive schema for a configuration setting."""
66 key: str
67 scope: list[str] = field(
68 default_factory=lambda: ["global"]
69 ) # ["global", "tenant", "user"]
70 type: str = "string" # string, int, bool, json, secret, enum
71 default: Any = None
72 options: list[str] | dict[str, str] | None = None # For enums
73 label: str = ""
74 description: str | None = None
75 category: str = "General"
76 is_public: bool = False
77 validation_rules: str | None = None # e.g. "min:1|max:100"
80class SettingsRegistry:
81 """Registry for setting definitions."""
83 _definitions: dict[str, SettingDefinition] = {}
85 @classmethod
86 def register(cls, definition: SettingDefinition) -> Any:
87 cls._definitions[definition.key] = definition
89 @classmethod
90 def get(cls, key: str) -> SettingDefinition | None:
91 return cls._definitions.get(key)
93 @classmethod
94 def get_all(cls) -> list[SettingDefinition]:
95 return list(cls._definitions.values())
97 @classmethod
98 def get_for_scope(cls, scope: str) -> list[SettingDefinition]:
99 return list(filter(lambda d: scope in d.scope, cls._definitions.values()))
102@inject
103class SettingsService:
104 """Service for managing implementation and resolution of settings."""
106 def __init__(
107 self,
108 repository: _SettingsRepository,
109 event_bus: Any = None,
110 cache: CacheBackendProtocol | None = None,
111 ):
112 self.repo = repository
113 self.registry = SettingsRegistry()
114 self.event_bus = event_bus
115 self.cache = cache
117 async def get(self, key: str, context: Any | None = None) -> Any:
118 """
119 Resolve a setting value with cascading logic and caching.
120 """
121 definition = self.registry.get(key)
123 # 1. Try Cache if available
124 cache_key = self._get_cache_key(key, context)
125 if self.cache:
126 try:
127 res = await self.cache.get(cache_key)
128 if res.is_ok():
129 cached = res.unwrap()
130 if cached is not None:
131 return cached
132 except (RuntimeError, ValueError, OSError) as e:
133 logger.warning("Cache lookup failed for %s: %s", key, e)
135 # 2. Cascading Strategy:
136 resolved_val = await self._resolve_cascaded(key, context, definition)
138 # 3. Cache and return
139 if self.cache:
140 try:
141 await self.cache.set(cache_key, resolved_val, ttl=3600)
142 except (RuntimeError, ValueError, OSError):
143 logger.exception("Cache set failed")
145 return resolved_val
147 async def _resolve_cascaded(
148 self,
149 key: str,
150 context: Any | None,
151 definition: SettingDefinition | None,
152 ) -> Any:
153 # 1. Env Override (Highest Priority) - LEX_KEY_STYLE
154 env_val = self._get_from_env(key)
155 if env_val is not None:
156 return self._cast_value(env_val, definition)
158 # 2. Database: User Scope
159 if context:
160 user = getattr(context, "user", None) or getattr(
161 getattr(context, "state", None),
162 "user",
163 None,
164 )
165 # Prefer canonical `user_id` attribute on user objects
166 if user and hasattr(user, "user_id"):
167 user_val = await self._get_from_repo(key, "user", str(user.user_id))
168 if user_val is not None:
169 return self._cast_value(user_val, definition)
171 # 3. Database: Global Scope
172 global_val = await self._get_from_repo(key, "global", "system")
173 if global_val is not None:
174 return self._cast_value(global_val, definition)
176 # 4. YAML Config (LexigramConfig)
177 yaml_val = self._get_from_yaml_config(key)
178 if yaml_val is not None:
179 return yaml_val
181 # 5. Default from Registry Definition
182 if definition:
183 return definition.default
185 # 6. Fallback to ConfigRegistry for new ConfigSpecs
186 try:
187 from lexigram.admin.lib.di import get_admin_resolver
188 from lexigram.admin.settings.panel.registry import ConfigRegistry
190 resolver = get_admin_resolver(context)
191 registry = await resolver.resolve(ConfigRegistry)
192 node = registry.get_node(key)
193 if node:
194 return node.default
195 except ImportError:
196 _warn_registry_unavailable()
198 return None
200 def _get_from_env(self, key: str) -> str | None:
201 """Get value from environment variable (LEX_ADMIN__KEY__STYLE)."""
202 import os
204 env_key = f"LEX_ADMIN__{key.upper().replace('.', '__')}"
205 return os.environ.get(env_key)
207 def _get_from_yaml_config(self, key: str) -> Any:
208 """Get value from boot-time config via LexigramConfig.
210 Traverses the boot config dict using dot-separated keys.
211 """
212 try:
213 from lexigram.config import LexigramConfig
215 config = LexigramConfig.boot_config() # type: ignore[attr-defined]
216 config_dict = config.model_dump()
217 parts = key.split(".")
218 curr = config_dict
219 for p in parts:
220 if isinstance(curr, dict) and p in curr:
221 curr = curr[p]
222 else:
223 return None
224 return curr
225 except (ImportError, RuntimeError):
226 return None
228 def _get_cache_key(self, key: str, context: Any | None = None) -> str:
229 """Generate a unique cache key based on key and context."""
230 scope_suffix = "global"
231 if context:
232 user = getattr(context, "user", None) or getattr(
233 getattr(context, "state", None),
234 "user",
235 None,
236 )
237 # Use canonical `user_id` when present
238 if user and hasattr(user, "user_id"):
239 scope_suffix = f"user:{user.user_id}"
240 return f"settings:resolved:{key}:{scope_suffix}"
242 async def _get_from_repo(
243 self,
244 key: str,
245 scope: str,
246 scope_id: str,
247 ) -> str | None:
248 """Fetch raw value from repository using simple filters."""
249 try:
250 # Using simple filter-based find_one
251 setting = await self.repo.find_one(scope=scope, scope_id=scope_id, key=key)
252 return setting.value if setting else None
253 except BaseException:
254 logger.exception("Failed to fetch setting %s from repo", key)
255 return None
257 def _cast_value(self, value: str, definition: SettingDefinition | None) -> Any:
258 # ... (rest of the method unchanged)
259 if not definition:
260 return value
262 if definition.type == "bool":
263 return value.lower() == "true"
264 if definition.type == "int":
265 try:
266 return int(value)
267 except ValueError:
268 return 0
269 elif definition.type == "json":
270 from lexigram.serialization import loads_str
272 try:
273 return loads_str(value)
274 except (json.JSONDecodeError, ValueError, TypeError) as e:
275 logger.warning(
276 "Failed to parse JSON setting for %s: %s",
277 definition.key,
278 e,
279 )
280 return {}
282 return value
284 async def set(
285 self,
286 key: str,
287 value: Any,
288 scope: str = "global",
289 scope_id: str = "system",
290 context: Any | None = None,
291 ) -> Any:
292 """Persist a setting value with audit tracking and cache invalidation."""
294 # 1. Resolve Auditor
295 updated_by = None
296 if context:
297 user = getattr(context, "user", None) or getattr(
298 getattr(context, "state", None),
299 "user",
300 None,
301 )
302 if user:
303 # Prefer username, then canonical user_id, then fallback to str(user)
304 updated_by = getattr(
305 user,
306 "username",
307 getattr(user, "user_id", str(user)),
308 )
310 # 2. Type serialization
311 str_value = str(value)
312 if isinstance(value, bool):
313 str_value = "true" if value else "false"
314 elif isinstance(value, (dict, list)):
315 from lexigram.serialization import dumps_str
317 str_value = dumps_str(value)
319 definition = self.registry.get(key)
320 setting_type = "string"
321 is_sensitive = False
323 if definition:
324 setting_type = definition.type
325 is_sensitive = definition.type == "secret"
326 else:
327 # Fallback to ConfigRegistry
328 try:
329 from lexigram.admin.lib.di import get_admin_resolver
330 from lexigram.admin.settings.panel.registry import ConfigRegistry
332 resolver = get_admin_resolver(context)
333 config_registry = await resolver.resolve(
334 ConfigRegistry,
335 )
336 node = config_registry.get_node(key)
337 if node:
338 setting_type = map_config_node_type(node)
339 is_sensitive = setting_type == "secret"
340 except ImportError:
341 _warn_registry_unavailable()
343 try:
344 existing = await self.repo.find_one(scope=scope, scope_id=scope_id, key=key)
346 if existing:
347 existing.value = str_value
348 existing.updated_at = datetime.now(UTC)
349 existing.updated_by = updated_by
350 await self.repo.update(existing)
351 else:
352 new_setting = SystemSetting(
353 scope=scope,
354 scope_id=scope_id,
355 key=key,
356 value=str_value,
357 type=setting_type,
358 is_sensitive=is_sensitive,
359 updated_by=updated_by,
360 )
361 await self.repo.create(new_setting)
363 # 3. Invalidate Cache
364 if self.cache:
365 await self.cache.delete(f"settings:resolved:{key}:global")
366 if scope == "user":
367 await self.cache.delete(f"settings:resolved:{key}:user:{scope_id}")
369 # Fire Event
370 if self.event_bus:
371 try:
372 await self.event_bus.publish(
373 SettingUpdated(
374 key=key,
375 value=str_value,
376 scope=scope,
377 scope_id=scope_id,
378 ),
379 )
380 except BaseException:
381 logger.exception("Failed to publish SettingUpdated event")
383 except BaseException:
384 logger.exception("Failed to save setting %s", key)
385 raise
387 async def get_all_resolved(self, context: Any | None = None) -> dict[str, Any]:
388 """Return a dictionary of all resolved settings for the current context."""
389 settings = {}
390 for definition in self.registry.get_all():
391 settings[definition.key] = await self.get(definition.key, context)
392 return settings
394 async def boot(self, container: ContainerResolverProtocol | None = None) -> None:
395 """Initialize settings resources and register as Config Store."""
396 # Register as a store for ConfigRegistry
397 from lexigram.admin.lib.di import get_admin_resolver
398 from lexigram.admin.settings.panel.registry import ConfigRegistry, StoreBase
400 resolver = get_admin_resolver(container)
401 registry = await resolver.resolve(ConfigRegistry)
403 class SettingsServiceStore(StoreBase):
404 def __init__(self, service: SettingsService):
405 self.service = service
407 async def get(self, key: str, default: Any = None) -> Any:
408 val = await self.service.get(key)
409 return val if val is not None else default
411 async def set(self, key: str, value: Any) -> None:
412 await self.service.set(key, value)
414 # Register store named 'app' (matches ConfigSpec category 'app' if we map it)
415 # But ConfigRegistry uses store_name="default" by default.
416 # We should register it as 'db' or override 'default'?
417 # For now, let's register as 'default' to take over persistence globally
418 # (since MemoryStore is useless in production)
419 registry.register_store("default", SettingsServiceStore(self))
421 logger.info("✓ SettingsService registered as default ConfigRegistry store")