Coverage for src/lexigram/admin/core/registry.py: 0%

1from __future__ import annotations 

2 

3import importlib 

4import inspect 

5import pkgutil 

6from typing import TYPE_CHECKING, Any, Self 

7 

8from lexigram.admin.config import AdminConfig 

9from lexigram.admin.models.provider_models import Command 

10from lexigram.contracts.exceptions import ConfigurationError 

11from lexigram.logging import get_logger 

12 

13if TYPE_CHECKING: 

14 from lexigram.admin.resources.base import Resource 

15 from lexigram.contracts.core.di import ContainerRegistrarProtocol 

16 

17logger = get_logger(__name__) 

18 

19 

20class AdminRegistry: 

21 """Standalone registry for admin resources and controllers.""" 

22 

23 def __init__(self, config: AdminConfig | None = None): 

24 self._config = config 

25 self._resources: dict[str, Any] = {} 

26 self._deferred_resources: dict[str, type] = {} 

27 self._controllers: list[Any] = [] 

28 self._commands: list[Command] = [] 

29 self._mounted: bool = False 

30 

31 @property 

32 def resources(self) -> dict[str, Any]: 

33 return self._resources 

34 

35 @property 

36 def controllers(self) -> list[Any]: 

37 return self._controllers 

38 

39 @property 

40 def commands(self) -> list[Command]: 

41 return self._commands 

42 

43 def register_resource( 

44 self, 

45 resource: type | Resource, 

46 *, 

47 name: str | None = None, 

48 group: str | None = None, 

49 ) -> Self: 

50 if self._mounted: 

51 raise ConfigurationError(message="Cannot register resources after mounting") 

52 

53 is_class = isinstance(resource, type) 

54 resource_name = name or self._extract_resource_name(resource, is_class) 

55 resource_group = group or self._extract_resource_group(resource, is_class) 

56 

57 if is_class: 

58 self._deferred_resources[resource_name] = resource # type: ignore[assignment] 

59 logger.debug("Deferred resource registration: %s", resource_name) 

60 else: 

61 self._resources[resource_name] = resource 

62 logger.debug("Registered resource: %s", resource_name) 

63 

64 if resource_group: 

65 self._assign_resource_to_group( 

66 resource_name, resource_group, resource, is_class 

67 ) 

68 

69 return self 

70 

71 @staticmethod 

72 def _extract_resource_name(resource: type | Any, is_class: bool) -> str: 

73 name = getattr(resource, "name", None) 

74 if not name: 

75 cfg = getattr(resource, "config", None) 

76 if cfg is not None: 

77 name = getattr(cfg, "_name", None) 

78 if not name: 

79 cls = resource if is_class else resource.__class__ 

80 name = cls.__name__.lower().replace("resource", "") 

81 return name 

82 

83 @staticmethod 

84 def _extract_resource_group(resource: type | Any, _is_class: bool) -> str | None: 

85 group = getattr(resource, "group", None) 

86 if not group: 

87 cfg = getattr(resource, "config", None) 

88 if cfg is not None: 

89 group = getattr(cfg, "_group", None) 

90 return group 

91 

92 def _assign_resource_to_group( 

93 self, 

94 resource_name: str, 

95 group_key: str, 

96 resource: type | Any, 

97 is_class: bool, 

98 ) -> None: 

99 nav_groups = self._config.navigation_groups if self._config else {} 

100 

101 if group_key not in nav_groups: 

102 cfg = getattr(resource, "config", None) 

103 group_label = None 

104 group_icon = None 

105 group_order = 100 

106 if cfg is not None: 

107 group_label = getattr(cfg, "_group_label", None) 

108 group_icon = getattr(cfg, "_group_icon", None) 

109 group_order = getattr(cfg, "_group_order", None) or 100 

110 

111 from lexigram.admin.config import AdminNavigationGroup 

112 

113 nav_groups[group_key] = AdminNavigationGroup( 

114 label=group_label or group_key.replace("_", " ").title(), 

115 icon=group_icon, 

116 order=group_order, 

117 ) 

118 logger.debug("Auto-created navigation group: %s", group_key) 

119 

120 nav_group = nav_groups[group_key] 

121 if resource_name not in nav_group.resources: 

122 nav_group.resources.append(resource_name) 

123 

124 def register_many(self, *resources: type | Resource) -> Self: 

125 for resource in resources: 

126 self.register_resource(resource) 

127 return self 

128 

129 def register_command(self, command: Any) -> Self: 

130 if isinstance(command, dict): 

131 cmd = Command( 

132 label=command.get("label", ""), 

133 href=command.get("href", ""), 

134 icon=command.get("icon", ""), 

135 shortcut=command.get("shortcut", ""), 

136 ) 

137 elif hasattr(command, "label"): 

138 cmd = Command( 

139 label=getattr(command, "label", ""), 

140 href=getattr(command, "href", ""), 

141 icon=getattr(command, "icon", ""), 

142 shortcut=getattr(command, "shortcut", ""), 

143 ) 

144 else: 

145 raise ValueError(f"Invalid command type: {type(command)}") 

146 

147 self._commands.append(cmd) 

148 return self 

149 

150 def register_controller(self, controller: type | Any) -> Self: 

151 if self._mounted: 

152 raise ConfigurationError( 

153 message="Cannot register controllers after mounting", 

154 ) 

155 

156 self._controllers.append(controller) 

157 logger.debug( 

158 "Registered controller: %s", 

159 getattr(controller, "__name__", str(controller)), 

160 ) 

161 return self 

162 

163 def discover_resources( 

164 self, 

165 package: str, 

166 container: ContainerRegistrarProtocol, 

167 ) -> Self: 

168 from lexigram.admin.resources.base import Resource 

169 

170 count = 0 

171 for cls in self._scan_package(package, Resource): 

172 container.transient(cls, cls) 

173 self.register_resource(cls) 

174 count += 1 

175 

176 logger.info("Discovered %d admin resources in %s", count, package) 

177 return self 

178 

179 def discover_controllers( 

180 self, 

181 package: str, 

182 container: ContainerRegistrarProtocol, 

183 ) -> Self: 

184 from lexigram.admin.controllers import AdminController 

185 

186 count = 0 

187 for cls in self._scan_package(package, AdminController): 

188 self.register_controller(cls) 

189 count += 1 

190 

191 logger.info("Discovered %d admin controllers in %s", count, package) 

192 return self 

193 

194 @staticmethod 

195 def _scan_package(package: str, base_class: type) -> list[type]: 

196 found: list[type] = [] 

197 seen: set[type] = set() 

198 

199 mod = importlib.import_module(package) 

200 pkg_path = getattr(mod, "__path__", None) 

201 if pkg_path is None: 

202 return found 

203 

204 for _importer, modname, _ispkg in pkgutil.walk_packages( 

205 pkg_path, prefix=package + "." 

206 ): 

207 try: 

208 submod = importlib.import_module(modname) 

209 except (ImportError, ModuleNotFoundError, AttributeError, TypeError) as exc: 

210 logger.debug("Skipping unimportable module: %s (%s)", modname, exc) 

211 continue 

212 

213 for attr_name in dir(submod): 

214 obj = getattr(submod, attr_name, None) 

215 if ( 

216 obj is not None 

217 and inspect.isclass(obj) 

218 and issubclass(obj, base_class) 

219 and obj is not base_class 

220 and not inspect.isabstract(obj) 

221 and obj not in seen 

222 ): 

223 seen.add(obj) 

224 found.append(obj) 

225 

226 return found