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

1"""Action registry: types, protocols, and ActionRegistry for admin actions.""" 

2 

3from __future__ import annotations 

4 

5from abc import ABC, abstractmethod 

6from dataclasses import dataclass, field 

7from enum import Enum 

8from typing import TYPE_CHECKING, Any, Protocol, TypeVar, runtime_checkable 

9 

10from lexigram.result import Ok, Result 

11 

12if TYPE_CHECKING: 

13 from collections.abc import Callable 

14 

15 from lexigram.admin.exceptions import ( 

16 AdminValidationError, 

17 ) 

18 from lexigram.contracts.admin.action_hooks import ActionHookProtocol 

19 

20T = TypeVar("T") 

21 

22 

23class ActionType(str, Enum): 

24 """Types of admin actions.""" 

25 

26 SINGLE = "single" # Operates on a single resource 

27 BULK = "bulk" # Operates on multiple resources 

28 GLOBAL = "global" # Not tied to resources 

29 

30 

31class ActionExecutionMode(str, Enum): 

32 """How an action should be executed.""" 

33 

34 SYNC = "sync" # Execute immediately 

35 ASYNC = "async" # Execute as background task 

36 CONFIRM = "confirm" # Require user confirmation first 

37 

38 

39@dataclass 

40class ActionConfig: 

41 """Configuration for an admin action.""" 

42 

43 name: str 

44 label: str 

45 description: str = "" 

46 icon: str | None = None 

47 

48 # Action type 

49 action_type: ActionType = ActionType.SINGLE 

50 execution_mode: ActionExecutionMode = ActionExecutionMode.SYNC 

51 

52 # Authorization 

53 permission: str | None = None 

54 

55 # Confirmation 

56 confirm_message: str | None = None 

57 confirm_style: str = "warning" # info, warning, danger 

58 

59 # UI 

60 button_variant: str = "secondary" 

61 show_in_list: bool = True 

62 show_in_detail: bool = True 

63 

64 # Bulk specific 

65 min_selection: int = 1 

66 max_selection: int | None = None 

67 

68 # Form for action parameters 

69 has_form: bool = False 

70 form_schema: dict[str, Any] | None = None 

71 

72 

73@dataclass 

74class ActionContext: 

75 """Context for action execution.""" 

76 

77 user: Any 

78 resource_name: str 

79 action_name: str 

80 

81 # Target(s) 

82 record_id: Any | None = None 

83 record_ids: list[Any] = field(default_factory=list) 

84 

85 # Form data 

86 parameters: dict[str, Any] = field(default_factory=dict) 

87 

88 # Additional context 

89 request_id: str | None = None 

90 metadata: dict[str, Any] = field(default_factory=dict) 

91 

92 @property 

93 def is_bulk(self) -> bool: 

94 """Check if this is a bulk action.""" 

95 return bool( 

96 len(self.record_ids) > 1 or (not self.record_id and self.record_ids) 

97 ) 

98 

99 @property 

100 def target_count(self) -> int: 

101 """Get the number of target records.""" 

102 if self.record_ids: 

103 return len(self.record_ids) 

104 return 1 if self.record_id else 0 

105 

106 

107@dataclass 

108class ActionResult: 

109 """Result of a successful action execution.""" 

110 

111 message: str 

112 data: Any = None 

113 

114 # For bulk actions 

115 successful_count: int = 0 

116 failed_count: int = 0 

117 failures: list[tuple[Any, str]] = field(default_factory=list) # (id, error_msg) 

118 

119 # For async actions 

120 task_id: str | None = None 

121 

122 # Redirect 

123 redirect_url: str | None = None 

124 

125 # Refresh target 

126 refresh_target: str | None = None 

127 

128 @classmethod 

129 def bulk( 

130 cls, 

131 message: str, 

132 successful_count: int, 

133 failed_count: int, 

134 failures: list[tuple[Any, str]] | None = None, 

135 ) -> ActionResult: 

136 """Create a bulk action result.""" 

137 return cls( 

138 message=message, 

139 successful_count=successful_count, 

140 failed_count=failed_count, 

141 failures=failures or [], 

142 ) 

143 

144 @classmethod 

145 def async_started( 

146 cls, 

147 task_id: str, 

148 message: str = "Action started", 

149 ) -> ActionResult: 

150 """Create an async action started result.""" 

151 return cls( 

152 message=message, 

153 task_id=task_id, 

154 ) 

155 

156 

157@runtime_checkable 

158class ActionHandler(Protocol): 

159 """Protocol for action handlers.""" 

160 

161 async def execute(self, context: ActionContext) -> ActionResult: ... 

162 

163 

164@runtime_checkable 

165class ActionValidator(Protocol): 

166 """Protocol for action validators.""" 

167 

168 async def validate( 

169 self, 

170 context: ActionContext, 

171 config: ActionConfig, 

172 ) -> Result[None, AdminValidationError]: ... 

173 

174 

175class AbstractActionHandler(ABC): 

176 """Base class for action handlers.""" 

177 

178 def __init__(self) -> None: 

179 """Initialize lifecycle hook collections.""" 

180 self.before_hooks: list[ActionHookProtocol] = [] 

181 self.after_hooks: list[ActionHookProtocol] = [] 

182 self.failure_hooks: list[ActionHookProtocol] = [] 

183 

184 def register_hooks( 

185 self, 

186 *, 

187 before: list[ActionHookProtocol] | None = None, 

188 after: list[ActionHookProtocol] | None = None, 

189 failure: list[ActionHookProtocol] | None = None, 

190 ) -> None: 

191 """Register lifecycle hooks on this handler. 

192 

193 Args: 

194 before: Hooks run before the action body. 

195 after: Hooks run after successful execution. 

196 failure: Hooks run when the action fails. 

197 """ 

198 if before: 

199 self.before_hooks.extend(before) 

200 if after: 

201 self.after_hooks.extend(after) 

202 if failure: 

203 self.failure_hooks.extend(failure) 

204 

205 @abstractmethod 

206 async def execute(self, context: ActionContext) -> ActionResult: 

207 """Execute the action.""" 

208 ... 

209 

210 async def validate( 

211 self, 

212 context: ActionContext, 

213 config: ActionConfig, 

214 ) -> Result[None, AdminValidationError]: 

215 """Validate action parameters. Override for custom validation.""" 

216 return Ok(None) 

217 

218 async def on_success(self, context: ActionContext, result: ActionResult) -> None: 

219 """Called after successful execution. Override for post-processing.""" 

220 

221 async def on_failure(self, context: ActionContext, error: Exception) -> None: 

222 """Called after failed execution. Override for error handling.""" 

223 

224 

225class FunctionActionHandler(AbstractActionHandler): 

226 """Action handler that wraps a function.""" 

227 

228 def __init__( 

229 self, 

230 func: Callable[[ActionContext], Any], 

231 is_async: bool = True, 

232 ): 

233 super().__init__() 

234 self.func = func 

235 self.is_async = is_async 

236 

237 async def execute(self, context: ActionContext) -> ActionResult: 

238 """Execute the wrapped function.""" 

239 if self.is_async: 

240 result = await self.func(context) 

241 else: 

242 result = self.func(context) 

243 

244 if isinstance(result, ActionResult): 

245 return result 

246 

247 return ActionResult(message="Action completed", data=result) 

248 

249 

250class ActionRegistry: 

251 """Registry for admin actions.""" 

252 

253 def __init__(self) -> None: 

254 self._actions: dict[str, dict[str, tuple[ActionConfig, ActionHandler]]] = {} 

255 # resource_name -> action_name -> (config, handler) 

256 

257 self._global_actions: dict[str, tuple[ActionConfig, ActionHandler]] = {} 

258 

259 def register( 

260 self, 

261 resource_name: str, 

262 config: ActionConfig, 

263 handler: ActionHandler | Callable[[ActionContext], Any], 

264 ) -> None: 

265 """Register an action for a resource. 

266 

267 Args: 

268 resource_name: Resource this action applies to 

269 config: Action configuration 

270 handler: Action handler or callable 

271 """ 

272 if resource_name not in self._actions: 

273 self._actions[resource_name] = {} 

274 

275 if callable(handler) and not isinstance(handler, ActionHandler): 

276 handler = FunctionActionHandler(handler) 

277 

278 self._actions[resource_name][config.name] = (config, handler) 

279 

280 def register_global( 

281 self, 

282 config: ActionConfig, 

283 handler: ActionHandler | Callable[[ActionContext], Any], 

284 ) -> None: 

285 """Register a global action (not tied to a resource). 

286 

287 Args: 

288 config: Action configuration 

289 handler: Action handler or callable 

290 """ 

291 if callable(handler) and not isinstance(handler, ActionHandler): 

292 handler = FunctionActionHandler(handler) 

293 

294 self._global_actions[config.name] = (config, handler) 

295 

296 def get( 

297 self, 

298 resource_name: str, 

299 action_name: str, 

300 ) -> tuple[ActionConfig, ActionHandler] | None: 

301 """Get an action by resource and name.""" 

302 if resource_name in self._actions: 

303 return self._actions[resource_name].get(action_name) 

304 return None 

305 

306 def get_global(self, action_name: str) -> tuple[ActionConfig, ActionHandler] | None: 

307 """Get a global action by name.""" 

308 return self._global_actions.get(action_name) 

309 

310 def get_actions_for_resource( 

311 self, 

312 resource_name: str, 

313 action_type: ActionType | None = None, 

314 ) -> list[ActionConfig]: 

315 """Get all actions for a resource. 

316 

317 Args: 

318 resource_name: Resource name 

319 action_type: Optional filter by action type 

320 

321 Returns: 

322 List of action configurations 

323 """ 

324 if resource_name not in self._actions: 

325 return [] 

326 

327 configs = [x[0] for x in self._actions[resource_name].values()] 

328 

329 if action_type: 

330 configs = list(filter(lambda c: c.action_type == action_type, configs)) 

331 

332 return configs 

333 

334 def get_all_global_actions(self) -> list[ActionConfig]: 

335 """Get all global actions.""" 

336 return [x[0] for x in self._global_actions.values()] 

337 

338 

339__all__ = [ 

340 "AbstractActionHandler", 

341 "ActionConfig", 

342 "ActionContext", 

343 "ActionExecutionMode", 

344 "ActionHandler", 

345 "ActionRegistry", 

346 "ActionResult", 

347 "ActionType", 

348 "ActionValidator", 

349 "FunctionActionHandler", 

350]