Coverage for src/lexigram/admin/settings/panel/registry.py: 31%

1"""Registry and persistence for configuration.""" 

2 

3from __future__ import annotations 

4 

5import os 

6from typing import TYPE_CHECKING, Any 

7 

8if TYPE_CHECKING: 

9 from lexigram.admin.settings.panel.nodes import AbstractConfigNode, ConfigSpec 

10 

11__all__ = [ 

12 "ConfigRegistry", 

13 "EnvStore", 

14 "MemoryStore", 

15 "StoreBase", 

16] 

17 

18 

19class StoreBase: 

20 """Interface for configuration persistence.""" 

21 

22 async def get( 

23 self, key: str, default: Any = None, tenant_id: str | None = None 

24 ) -> Any: 

25 """Retrieve a value by key.""" 

26 return default 

27 

28 async def set(self, key: str, value: Any, tenant_id: str | None = None) -> None: 

29 """Persist a value by key.""" 

30 

31 

32class EnvStore(StoreBase): 

33 """Read-only store for environment variables.""" 

34 

35 async def get( 

36 self, key: str, default: Any = None, tenant_id: str | None = None 

37 ) -> Any: 

38 """Read a value from environment variables. 

39 

40 Converts internal dot-notation keys (e.g. ``app.db.url``) to 

41 ``SCREAMING_SNAKE_CASE`` environment variable names. 

42 """ 

43 env_key = key.upper().replace(".", "_") 

44 return os.environ.get(env_key, default) 

45 

46 

47class MemoryStore(StoreBase): 

48 """In-memory store for testing.""" 

49 

50 def __init__(self) -> None: 

51 self._data: dict[str, Any] = {} 

52 

53 async def get( 

54 self, key: str, default: Any = None, tenant_id: str | None = None 

55 ) -> Any: 

56 """Retrieve a value from the in-memory store.""" 

57 return self._data.get(key, default) 

58 

59 async def set(self, key: str, value: Any, tenant_id: str | None = None) -> None: 

60 """Persist a value to the in-memory store.""" 

61 self._data[key] = value 

62 

63 

64class ConfigRegistry: 

65 """Central registry for specs and stores.""" 

66 

67 def __init__(self) -> None: 

68 self._specs: dict[str, type[ConfigSpec]] = {} 

69 self._stores: dict[str, StoreBase] = { 

70 "env": EnvStore(), 

71 "default": MemoryStore(), 

72 } 

73 

74 def register_store(self, name: str, store: StoreBase) -> None: 

75 """Register a configuration store.""" 

76 self._stores[name] = store 

77 

78 def register_spec(self, spec: type[ConfigSpec]) -> None: 

79 """Register a spec, grouped in the sidebar under its ``package_source``.""" 

80 if spec.namespace in self._specs: 

81 return 

82 self._specs[spec.namespace] = spec 

83 

84 def get_package_sources(self) -> list[str]: 

85 """Return distinct package sources among specs with editable nodes, sorted.""" 

86 return sorted( 

87 {spec.package_source for spec in self._specs.values() if spec.get_nodes()} 

88 ) 

89 

90 def get_specs_by_package(self, package_source: str) -> list[type[ConfigSpec]]: 

91 """Get all registered specs for a package source that have editable nodes.""" 

92 return [ 

93 spec 

94 for spec in self._specs.values() 

95 if spec.package_source == package_source and spec.get_nodes() 

96 ] 

97 

98 def get_spec(self, namespace: str) -> type[ConfigSpec] | None: 

99 """Get a registered spec by namespace.""" 

100 return self._specs.get(namespace) 

101 

102 def has_store(self, name: str) -> bool: 

103 """Check whether a store is registered.""" 

104 return name in self._stores 

105 

106 @classmethod 

107 def with_defaults(cls) -> ConfigRegistry: 

108 """Build a registry pre-populated with all built-in bound specs.""" 

109 registry = cls() 

110 from lexigram.admin.settings.panel import ( 

111 register_branding_spec, 

112 register_cache_spec, 

113 register_deployment_spec, 

114 register_features_spec, 

115 register_i18n_spec, 

116 register_profiler_spec, 

117 register_rate_limit_spec, 

118 register_rbac_spec, 

119 register_security_spec, 

120 ) 

121 

122 register_branding_spec(registry) 

123 register_cache_spec(registry) 

124 register_security_spec(registry) 

125 register_features_spec(registry) 

126 register_i18n_spec(registry) 

127 register_profiler_spec(registry) 

128 register_rate_limit_spec(registry) 

129 register_rbac_spec(registry) 

130 register_deployment_spec(registry) 

131 return registry 

132 

133 def get_node(self, full_key: str) -> AbstractConfigNode | None: 

134 """Get a ConfigNode by its full key (namespace.key).""" 

135 if "." not in full_key: 

136 return None 

137 

138 # Try to find the longest matching namespace 

139 for ns, spec in self._specs.items(): 

140 if full_key.startswith(ns + "."): 

141 node_key = full_key[len(ns) + 1 :] 

142 nodes = spec.get_nodes() 

143 if node_key in nodes: 

144 return nodes[node_key] 

145 return None 

146 

147 async def get_values( 

148 self, 

149 namespace: str, 

150 store_name: str = "default", 

151 tenant_id: str | None = None, 

152 ) -> dict[str, Any]: 

153 """Load current values for a spec from a store.""" 

154 spec = self._specs.get(namespace) 

155 if not spec: 

156 return {} 

157 

158 store = self._stores.get(store_name, self._stores["default"]) 

159 values = {} 

160 for key, node in spec.get_nodes().items(): 

161 full_key = f"{namespace}.{key}" 

162 raw_val = await store.get(full_key, node.default, tenant_id=tenant_id) 

163 values[key] = node.validate(raw_val) 

164 return values 

165 

166 async def save_values( 

167 self, 

168 namespace: str, 

169 values: dict[str, Any], 

170 store_name: str = "default", 

171 tenant_id: str | None = None, 

172 ) -> None: 

173 """Save values for a spec to a store, skipping readonly nodes.""" 

174 spec = self._specs.get(namespace) 

175 if not spec: 

176 return 

177 

178 store = self._stores.get(store_name, self._stores["default"]) 

179 nodes = spec.get_nodes() 

180 for key, value in values.items(): 

181 if key in nodes and not nodes[key].readonly: 

182 full_key = f"{namespace}.{key}" 

183 validated = nodes[key].validate(value) 

184 await store.set(full_key, validated, tenant_id=tenant_id)