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

1from __future__ import annotations 

2 

3from typing import TYPE_CHECKING, Any 

4 

5from lexigram.admin.sql_dialect import is_postgres, now_expr 

6from lexigram.contracts.tenancy.protocols import TenantConfigProviderProtocol 

7from lexigram.logging import get_logger 

8from lexigram.serialization import dumps_str, loads_str 

9 

10if TYPE_CHECKING: 

11 from lexigram.contracts.data import DatabaseProviderProtocol 

12 

13logger = get_logger(__name__) 

14 

15KEY_PREFIX = "admin_ui." 

16 

17DEFAULT_SETTINGS: dict[str, Any] = { 

18 "site_name": "Lexigram Admin", 

19 "primary_color": "#6b7280", 

20 "logo_url": "", 

21 "favicon_url": "", 

22 "dark_mode": "system", 

23} 

24 

25 

26_TABLE = "tenant_configs" 

27 

28_CREATE_SQL_POSTGRES = f""" 

29CREATE TABLE IF NOT EXISTS {_TABLE} ( 

30 tenant_id VARCHAR(255) NOT NULL, 

31 key VARCHAR(255) NOT NULL, 

32 value TEXT NOT NULL, 

33 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), 

34 updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), 

35 PRIMARY KEY (tenant_id, key) 

36) 

37""" 

38 

39_CREATE_SQL_SQLITE = f""" 

40CREATE TABLE IF NOT EXISTS {_TABLE} ( 

41 tenant_id VARCHAR(255) NOT NULL, 

42 key VARCHAR(255) NOT NULL, 

43 value TEXT NOT NULL, 

44 created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, 

45 updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, 

46 PRIMARY KEY (tenant_id, key) 

47) 

48""" 

49 

50 

51class AdminSettingsDbProvider(TenantConfigProviderProtocol): 

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

53 

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

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

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

57 """ 

58 

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

60 self._db = db 

61 self._initialized = False 

62 

63 async def _ensure_table(self) -> None: 

64 if self._initialized: 

65 return 

66 try: 

67 create_sql = ( 

68 _CREATE_SQL_POSTGRES if is_postgres(self._db) else _CREATE_SQL_SQLITE 

69 ) 

70 await self._db.execute(create_sql, []) 

71 self._initialized = True 

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

73 except Exception: 

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

75 raise 

76 

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

78 await self._ensure_table() 

79 result = await self._db.execute( 

80 f"SELECT value FROM {_TABLE} WHERE tenant_id = ? AND key = ?", # noqa: S608 — table name is module constant "tenant_configs", never user input 

81 [tenant_id, key], 

82 ) 

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

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

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

86 return None 

87 

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

89 await self._ensure_table() 

90 result = await self._db.execute( 

91 f"SELECT key, value FROM {_TABLE} WHERE tenant_id = ?", # noqa: S608 — table name is module constant "tenant_configs", never user input 

92 [tenant_id], 

93 ) 

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

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

96 k = row.get("key") 

97 v = row.get("value") 

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

99 rows[k] = loads_str(v) 

100 return rows 

101 

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

103 await self._ensure_table() 

104 await self._db.execute( 

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

106 VALUES (?, ?, ?) 

107 ON CONFLICT (tenant_id, key) 

108 DO UPDATE SET value = excluded.value, updated_at = {now_expr(self._db)}""", # noqa: S608 — table name is module constant, now_expr yields fixed NOW()/CURRENT_TIMESTAMP 

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

110 ) 

111 

112 

113class AdminSettingsService: 

114 def __init__( 

115 self, 

116 config_provider: TenantConfigProviderProtocol | None = None, 

117 ) -> None: 

118 self._provider = config_provider 

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

120 

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

122 return f"{KEY_PREFIX}{name}" 

123 

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

125 if self._provider is None: 

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

127 name 

128 ) 

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

130 if raw is not None: 

131 return raw 

132 return DEFAULT_SETTINGS.get(name) 

133 

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

135 if self._provider is None: 

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

137 return 

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

139 

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

141 merged = dict(DEFAULT_SETTINGS) 

142 if self._provider is not None: 

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

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

145 if k.startswith(KEY_PREFIX): 

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

147 else: 

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

149 merged.update(tenant_data) 

150 return merged 

151 

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

153 if self._provider is None: 

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

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

156 ) 

157 return 

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

159 if name in DEFAULT_SETTINGS: 

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

161 

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

163 key = f"widgets.{user_id}" 

164 if self._provider is None: 

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

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

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

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

169 

170 async def set_widget_prefs( 

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

172 ) -> None: 

173 key = f"widgets.{user_id}" 

174 if self._provider is None: 

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

176 return 

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

178 

179 

180async def resolve_admin_settings_service( 

181 container: Any, 

182) -> AdminSettingsService | None: 

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

184 

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

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

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

188 fall back to client-side defaults. 

189 

190 Args: 

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

192 

193 Returns: 

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

195 """ 

196 try: 

197 from lexigram.contracts.data import DatabaseProviderProtocol 

198 

199 try: 

200 db_provider = await container.resolve( 

201 DatabaseProviderProtocol, 

202 bypass_visibility=True, 

203 ) 

204 except TypeError: 

205 db_provider = await container.resolve(DatabaseProviderProtocol) 

206 return AdminSettingsService( 

207 config_provider=AdminSettingsDbProvider(db=db_provider) 

208 ) 

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

210 logger.exception( 

211 "admin.settings_service_resolve_failed", 

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

213 ) 

214 return None 

215 

216 

217__all__ = [ 

218 "DEFAULT_SETTINGS", 

219 "AdminSettingsDbProvider", 

220 "AdminSettingsService", 

221 "resolve_admin_settings_service", 

222]