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

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(self, key: str, default: Any = None) -> Any: 

23 """Retrieve a value by key.""" 

24 return default 

25 

26 async def set(self, key: str, value: Any) -> None: 

27 """Persist a value by key.""" 

28 

29 

30class EnvStore(StoreBase): 

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

32 

33 async def get(self, key: str, default: Any = None) -> Any: 

34 """Read a value from environment variables. 

35 

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

37 ``SCREAMING_SNAKE_CASE`` environment variable names. 

38 """ 

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

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

41 

42 

43class MemoryStore(StoreBase): 

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

45 

46 def __init__(self) -> None: 

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

48 

49 async def get(self, key: str, default: Any = None) -> Any: 

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

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

52 

53 async def set(self, key: str, value: Any) -> None: 

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

55 self._data[key] = value 

56 

57 

58class ConfigRegistry: 

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

60 

61 def __init__(self) -> None: 

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

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

64 "env": EnvStore(), 

65 "default": MemoryStore(), 

66 } 

67 self._category_map: dict[str, list[str]] = { 

68 "env": [], 

69 "app": [], 

70 "system": [], 

71 } 

72 

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

74 """Register a configuration store.""" 

75 self._stores[name] = store 

76 

77 def register_spec(self, category: str, spec: type[ConfigSpec]) -> None: 

78 """Register a spec under a category (env, admin, app).""" 

79 if spec.namespace in self._specs: 

80 return 

81 

82 self._specs[spec.namespace] = spec 

83 if category in self._category_map: 

84 self._category_map[category].append(spec.namespace) 

85 

86 def get_specs(self, category: str) -> list[type[ConfigSpec]]: 

87 """Get all registered specs for a category that have editable nodes.""" 

88 namespaces = self._category_map.get(category, []) 

89 return [ 

90 self._specs[ns] 

91 for ns in namespaces 

92 if ns in self._specs and self._specs[ns].get_nodes() 

93 ] 

94 

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

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

97 return self._specs.get(namespace) 

98 

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

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

101 return name in self._stores 

102 

103 @classmethod 

104 def with_defaults(cls) -> ConfigRegistry: 

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

106 registry = cls() 

107 from lexigram.admin.settings.panel import ( 

108 register_branding_spec, 

109 register_cache_spec, 

110 register_features_spec, 

111 register_i18n_spec, 

112 register_profiler_spec, 

113 register_rate_limit_spec, 

114 register_rbac_spec, 

115 register_security_spec, 

116 ) 

117 

118 register_branding_spec(registry) 

119 register_cache_spec(registry) 

120 register_security_spec(registry) 

121 register_features_spec(registry) 

122 register_i18n_spec(registry) 

123 register_profiler_spec(registry) 

124 register_rate_limit_spec(registry) 

125 register_rbac_spec(registry) 

126 return registry 

127 

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

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

130 if "." not in full_key: 

131 return None 

132 

133 # Try to find the longest matching namespace 

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

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

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

137 nodes = spec.get_nodes() 

138 if node_key in nodes: 

139 return nodes[node_key] 

140 return None 

141 

142 async def get_values( 

143 self, 

144 namespace: str, 

145 store_name: str = "default", 

146 ) -> dict[str, Any]: 

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

148 spec = self._specs.get(namespace) 

149 if not spec: 

150 return {} 

151 

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

153 values = {} 

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

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

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

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

158 return values 

159 

160 async def save_values( 

161 self, 

162 namespace: str, 

163 values: dict[str, Any], 

164 store_name: str = "default", 

165 ) -> None: 

166 """Save values for a spec to a store.""" 

167 spec = self._specs.get(namespace) 

168 if not spec: 

169 return 

170 

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

172 nodes = spec.get_nodes() 

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

174 if key in nodes: 

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

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

177 await store.set(full_key, validated)