Coverage for src / lexigram / admin / services / settings_service.py: 36%

1from __future__ import annotations 

2 

3from typing import TYPE_CHECKING, Any 

4 

5from lexigram.contracts.tenancy.protocols import TenantConfigProviderProtocol 

6from lexigram.logging import get_logger 

7from lexigram.serialization import dumps_str, loads_str 

8 

9if TYPE_CHECKING: 

10 from lexigram.contracts.data import DatabaseProviderProtocol 

11 

12logger = get_logger(__name__) 

13 

14KEY_PREFIX = "admin_ui." 

15 

16DEFAULT_SETTINGS: dict[str, Any] = { 

17 "site_name": "Lexigram Admin", 

18 "primary_color": "#6b7280", 

19 "logo_url": "", 

20 "favicon_url": "", 

21 "dark_mode": "system", 

22} 

23 

24 

25_TABLE = "tenant_configs" 

26 

27_CREATE_SQL = f""" 

28CREATE TABLE IF NOT EXISTS {_TABLE} ( 

29 tenant_id VARCHAR(255) NOT NULL, 

30 key VARCHAR(255) NOT NULL, 

31 value TEXT NOT NULL, 

32 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), 

33 updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), 

34 PRIMARY KEY (tenant_id, key) 

35) 

36""" 

37 

38 

39class AdminSettingsDbProvider(TenantConfigProviderProtocol): 

40 """DB-backed tenant config provider that auto-creates its table. 

41 

42 Follows the pattern used by other lexigram packages (admin sessions, 

43 resilience idempotency, etc.) — creates ``tenant_configs`` on first 

44 use via ``CREATE TABLE IF NOT EXISTS``. 

45 """ 

46 

47 def __init__(self, db: DatabaseProviderProtocol) -> None: 

48 self._db = db 

49 self._initialized = False 

50 

51 async def _ensure_table(self) -> None: 

52 if self._initialized: 

53 return 

54 try: 

55 await self._db.execute(_CREATE_SQL, []) 

56 self._initialized = True 

57 logger.info("Ensured %s table exists", _TABLE) 

58 except Exception: 

59 logger.exception("Failed to create %s table", _TABLE) 

60 raise 

61 

62 async def get_config(self, tenant_id: str, key: str) -> Any | None: 

63 await self._ensure_table() 

64 result = await self._db.execute( 

65 f"SELECT value FROM {_TABLE} WHERE tenant_id = ? AND key = ?", 

66 [tenant_id, key], 

67 ) 

68 if hasattr(result, "rows") and result.rows: 

69 raw = result.rows[0].get("value") 

70 return loads_str(raw) if raw is not None else None 

71 return None 

72 

73 async def get_all_config(self, tenant_id: str) -> dict[str, Any]: 

74 await self._ensure_table() 

75 result = await self._db.execute( 

76 f"SELECT key, value FROM {_TABLE} WHERE tenant_id = ?", 

77 [tenant_id], 

78 ) 

79 rows: dict[str, Any] = {} 

80 for row in getattr(result, "rows", []) or []: 

81 k = row.get("key") 

82 v = row.get("value") 

83 if k is not None and v is not None: 

84 rows[k] = loads_str(v) 

85 return rows 

86 

87 async def set_config(self, tenant_id: str, key: str, value: Any) -> None: 

88 await self._ensure_table() 

89 await self._db.execute( 

90 f"""INSERT INTO {_TABLE} (tenant_id, key, value) 

91 VALUES (?, ?, ?) 

92 ON CONFLICT (tenant_id, key) 

93 DO UPDATE SET value = EXCLUDED.value, updated_at = NOW()""", 

94 [tenant_id, key, dumps_str(value)], 

95 ) 

96 

97 

98class AdminSettingsService: 

99 def __init__( 

100 self, 

101 config_provider: TenantConfigProviderProtocol | None = None, 

102 ) -> None: 

103 self._provider = config_provider 

104 self._memory: dict[str, dict[str, Any]] = {} 

105 

106 def _key(self, name: str) -> str: 

107 return f"{KEY_PREFIX}{name}" 

108 

109 async def get(self, tenant_id: str, name: str) -> Any: 

110 if self._provider is None: 

111 return self._memory.get(tenant_id, {}).get(name) or DEFAULT_SETTINGS.get( 

112 name 

113 ) 

114 raw = await self._provider.get_config(tenant_id, self._key(name)) 

115 if raw is not None: 

116 return raw 

117 return DEFAULT_SETTINGS.get(name) 

118 

119 async def set(self, tenant_id: str, name: str, value: Any) -> None: 

120 if self._provider is None: 

121 self._memory.setdefault(tenant_id, {})[name] = value 

122 return 

123 await self._provider.set_config(tenant_id, self._key(name), value) 

124 

125 async def get_all(self, tenant_id: str) -> dict[str, Any]: 

126 merged = dict(DEFAULT_SETTINGS) 

127 if self._provider is not None: 

128 raw = await self._provider.get_all_config(tenant_id) 

129 for k, v in raw.items(): 

130 if k.startswith(KEY_PREFIX): 

131 merged[k[len(KEY_PREFIX) :]] = v 

132 else: 

133 tenant_data = self._memory.get(tenant_id, {}) 

134 merged.update(tenant_data) 

135 return merged 

136 

137 async def set_all(self, tenant_id: str, settings: dict[str, Any]) -> None: 

138 if self._provider is None: 

139 self._memory.setdefault(tenant_id, {}).update( 

140 {k: v for k, v in settings.items() if k in DEFAULT_SETTINGS} 

141 ) 

142 return 

143 for name, value in settings.items(): 

144 if name in DEFAULT_SETTINGS: 

145 await self._provider.set_config(tenant_id, self._key(name), value) 

146 

147 async def get_widget_prefs(self, tenant_id: str, user_id: str) -> dict[str, Any]: 

148 key = f"widgets.{user_id}" 

149 if self._provider is None: 

150 data = self._memory.get(tenant_id, {}).get(key) 

151 return data if isinstance(data, dict) else {} 

152 raw = await self._provider.get_config(tenant_id, self._key(key)) 

153 return raw if isinstance(raw, dict) else {} 

154 

155 async def set_widget_prefs( 

156 self, tenant_id: str, user_id: str, prefs: dict[str, Any] 

157 ) -> None: 

158 key = f"widgets.{user_id}" 

159 if self._provider is None: 

160 self._memory.setdefault(tenant_id, {})[key] = prefs 

161 return 

162 await self._provider.set_config(tenant_id, self._key(key), prefs) 

163 

164 

165async def resolve_admin_settings_service( 

166 container: Any, 

167) -> AdminSettingsService | None: 

168 """Build a DB-backed settings service from a DI container. 

169 

170 Mirrors the bundle's own construction: resolves the database provider 

171 and wires an :class:`AdminSettingsDbProvider` underneath. Returns 

172 ``None`` when the database provider is unavailable, so callers can 

173 fall back to client-side defaults. 

174 

175 Args: 

176 container: DI resolver (``ContainerResolverProtocol``). 

177 

178 Returns: 

179 An ``AdminSettingsService`` or ``None`` if it cannot be built. 

180 """ 

181 try: 

182 from lexigram.contracts.data import DatabaseProviderProtocol 

183 

184 try: 

185 db_provider = await container.resolve( 

186 DatabaseProviderProtocol, 

187 bypass_visibility=True, 

188 ) 

189 except TypeError: 

190 db_provider = await container.resolve(DatabaseProviderProtocol) 

191 return AdminSettingsService( 

192 config_provider=AdminSettingsDbProvider(db=db_provider) 

193 ) 

194 except Exception as exc: # noqa: BLE001 — non-fatal 

195 logger.exception( 

196 "admin.settings_service_resolve_failed", 

197 error=str(exc)[:300], 

198 ) 

199 return None 

200 

201 

202__all__ = [ 

203 "DEFAULT_SETTINGS", 

204 "AdminSettingsDbProvider", 

205 "AdminSettingsService", 

206 "resolve_admin_settings_service", 

207]