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

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 

19T = TypeVar("T") 

20 

21 

22class ActionType(str, Enum): 

23 """Types of admin actions.""" 

24 

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

26 BULK = "bulk" # Operates on multiple resources 

27 GLOBAL = "global" # Not tied to resources 

28 

29 

30class ActionExecutionMode(str, Enum): 

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

32 

33 SYNC = "sync" # Execute immediately 

34 ASYNC = "async" # Execute as background task 

35 CONFIRM = "confirm" # Require user confirmation first 

36 

37 

38@dataclass 

39class ActionConfig: 

40 """Configuration for an admin action.""" 

41 

42 name: str 

43 label: str 

44 description: str = "" 

45 icon: str | None = None 

46 

47 # Action type 

48 action_type: ActionType = ActionType.SINGLE 

49 execution_mode: ActionExecutionMode = ActionExecutionMode.SYNC 

50 

51 # Authorization 

52 permission: str | None = None 

53 

54 # Confirmation 

55 confirm_message: str | None = None 

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

57 

58 # UI 

59 button_variant: str = "secondary" 

60 show_in_list: bool = True 

61 show_in_detail: bool = True 

62 

63 # Bulk specific 

64 min_selection: int = 1 

65 max_selection: int | None = None 

66 

67 # Form for action parameters 

68 has_form: bool = False 

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

70 

71 

72@dataclass 

73class ActionContext: 

74 """Context for action execution.""" 

75 

76 user: Any 

77 resource_name: str 

78 action_name: str 

79 

80 # Target(s) 

81 record_id: Any | None = None 

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

83 

84 # Form data 

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

86 

87 # Additional context 

88 request_id: str | None = None 

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

90 

91 @property 

92 def is_bulk(self) -> bool: 

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

94 return bool( 

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

96 ) 

97 

98 @property 

99 def target_count(self) -> int: 

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

101 if self.record_ids: 

102 return len(self.record_ids) 

103 return 1 if self.record_id else 0 

104 

105 

106@dataclass 

107class ActionResult: 

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

109 

110 message: str 

111 data: Any = None 

112 

113 # For bulk actions 

114 successful_count: int = 0 

115 failed_count: int = 0 

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

117 

118 # For async actions 

119 task_id: str | None = None 

120 

121 # Redirect 

122 redirect_url: str | None = None 

123 

124 # Refresh target 

125 refresh_target: str | None = None 

126 

127 @classmethod 

128 def bulk( 

129 cls, 

130 message: str, 

131 successful_count: int, 

132 failed_count: int, 

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

134 ) -> ActionResult: 

135 """Create a bulk action result.""" 

136 return cls( 

137 message=message, 

138 successful_count=successful_count, 

139 failed_count=failed_count, 

140 failures=failures or [], 

141 ) 

142 

143 @classmethod 

144 def async_started( 

145 cls, 

146 task_id: str, 

147 message: str = "Action started", 

148 ) -> ActionResult: 

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

150 return cls( 

151 message=message, 

152 task_id=task_id, 

153 ) 

154 

155 

156@runtime_checkable 

157class ActionHandler(Protocol): 

158 """Protocol for action handlers.""" 

159 

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

161 

162 

163@runtime_checkable 

164class ActionValidator(Protocol): 

165 """Protocol for action validators.""" 

166 

167 async def validate( 

168 self, 

169 context: ActionContext, 

170 config: ActionConfig, 

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

172 

173 

174class AbstractActionHandler(ABC): 

175 """Base class for action handlers.""" 

176 

177 @abstractmethod 

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

179 """Execute the action.""" 

180 ... 

181 

182 async def validate( 

183 self, 

184 context: ActionContext, 

185 config: ActionConfig, 

186 ) -> Result[None, AdminValidationError]: 

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

188 return Ok(None) 

189 

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

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

192 

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

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

195 

196 

197class FunctionActionHandler(AbstractActionHandler): 

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

199 

200 def __init__( 

201 self, 

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

203 is_async: bool = True, 

204 ): 

205 self.func = func 

206 self.is_async = is_async 

207 

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

209 """Execute the wrapped function.""" 

210 if self.is_async: 

211 result = await self.func(context) 

212 else: 

213 result = self.func(context) 

214 

215 if isinstance(result, ActionResult): 

216 return result 

217 

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

219 

220 

221class ActionRegistry: 

222 """Registry for admin actions.""" 

223 

224 def __init__(self) -> None: 

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

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

227 

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

229 

230 def register( 

231 self, 

232 resource_name: str, 

233 config: ActionConfig, 

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

235 ) -> None: 

236 """Register an action for a resource. 

237 

238 Args: 

239 resource_name: Resource this action applies to 

240 config: Action configuration 

241 handler: Action handler or callable 

242 """ 

243 if resource_name not in self._actions: 

244 self._actions[resource_name] = {} 

245 

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

247 handler = FunctionActionHandler(handler) 

248 

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

250 

251 def register_global( 

252 self, 

253 config: ActionConfig, 

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

255 ) -> None: 

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

257 

258 Args: 

259 config: Action configuration 

260 handler: Action handler or callable 

261 """ 

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

263 handler = FunctionActionHandler(handler) 

264 

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

266 

267 def get( 

268 self, 

269 resource_name: str, 

270 action_name: str, 

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

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

273 if resource_name in self._actions: 

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

275 return None 

276 

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

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

279 return self._global_actions.get(action_name) 

280 

281 def get_actions_for_resource( 

282 self, 

283 resource_name: str, 

284 action_type: ActionType | None = None, 

285 ) -> list[ActionConfig]: 

286 """Get all actions for a resource. 

287 

288 Args: 

289 resource_name: Resource name 

290 action_type: Optional filter by action type 

291 

292 Returns: 

293 List of action configurations 

294 """ 

295 if resource_name not in self._actions: 

296 return [] 

297 

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

299 

300 if action_type: 

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

302 

303 return configs 

304 

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

306 """Get all global actions.""" 

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

308 

309 

310__all__ = [ 

311 "AbstractActionHandler", 

312 "ActionConfig", 

313 "ActionContext", 

314 "ActionExecutionMode", 

315 "ActionHandler", 

316 "ActionRegistry", 

317 "ActionResult", 

318 "ActionType", 

319 "ActionValidator", 

320 "FunctionActionHandler", 

321]