Coverage for src / lexigram / admin / services / action_executor.py: 25%
84 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +0800
1"""ActionExecutor for executing custom admin actions with validation and authorization.
3See action_registry for types, protocols, and ActionRegistry.
4"""
6from __future__ import annotations
8from collections.abc import Callable
9from typing import Any, Protocol, runtime_checkable
11from lexigram.admin.exceptions import (
12 AdminError,
13 AdminValidationError,
14 PermissionDeniedError,
15)
16from lexigram.admin.services.action_registry import (
17 ActionConfig,
18 ActionContext,
19 ActionExecutionMode,
20 ActionHandler,
21 ActionRegistry,
22 ActionResult,
23 ActionType,
24)
25from lexigram.contracts.admin.authorizer import AdminAuthorizerProtocol
26from lexigram.di.decorators import inject
27from lexigram.result import Err, Ok, Result
30@runtime_checkable
31class TaskScheduler(Protocol):
32 """Protocol for scheduling background tasks."""
34 async def schedule(
35 self,
36 task_name: str,
37 args: tuple[Any, ...],
38 kwargs: dict[str, Any],
39 ) -> str: ...
42@inject
43class ActionExecutor:
44 """Executes admin actions with validation and authorization.
46 Handles:
47 - Action lookup and validation
48 - Authorization checks
49 - Sync and async execution
50 - Result handling
51 - Real-time notification publishing via AdminEventHub
53 Example:
54 >>> executor = ActionExecutor(registry, authorizer)
55 >>> context = ActionContext(
56 ... user=current_user,
57 ... resource_name="users",
58 ... action_name="deactivate",
59 ... record_id=123,
60 ... )
61 >>> result = await executor.execute(context)
62 """
64 def __init__(
65 self,
66 registry: ActionRegistry,
67 authorizer: AdminAuthorizerProtocol | None = None,
68 task_scheduler: TaskScheduler | None = None,
69 event_hub: Any | None = None,
70 ):
71 """Initialize the action executor.
73 Args:
74 registry: Action registry
75 authorizer: Optional authorizer for permission checks
76 task_scheduler: Optional task scheduler for async actions
77 event_hub: Optional AdminEventHub for publishing real-time notifications
78 """
79 self.registry = registry
80 self.authorizer = authorizer
81 self.task_scheduler = task_scheduler
82 self.event_hub = event_hub
84 async def execute(
85 self,
86 context: ActionContext,
87 ) -> Result[
88 ActionResult, PermissionDeniedError | AdminValidationError | AdminError
89 ]:
90 """Execute an action.
92 Args:
93 context: Action execution context
95 Returns:
96 Result containing ActionResult or error
97 """
98 # Get action
99 action = self.registry.get(context.resource_name, context.action_name)
100 if not action:
101 action = self.registry.get_global(context.action_name)
103 if not action:
104 return Err(
105 AdminError(
106 message=f"Action '{context.action_name}' not found for resource '{context.resource_name}'",
107 ),
108 )
110 config, handler = action
112 # Authorization
113 if self.authorizer and config.permission:
114 can_execute = await self.authorizer.can_execute_action(
115 context.user,
116 context.resource_name,
117 context.action_name,
118 )
119 if not can_execute:
120 return Err(
121 PermissionDeniedError(
122 resource=context.resource_name,
123 action=context.action_name,
124 message=f"No permission to execute '{context.action_name}'",
125 ),
126 )
128 # Validate bulk constraints
129 if config.action_type == ActionType.BULK:
130 if context.target_count < config.min_selection:
131 return Err(
132 AdminValidationError(
133 message=f"Select at least {config.min_selection} item(s)",
134 errors={ # type: ignore[arg-type]
135 "selection": [
136 f"Minimum {config.min_selection} items required",
137 ],
138 },
139 ),
140 )
141 if config.max_selection and context.target_count > config.max_selection:
142 return Err(
143 AdminValidationError(
144 message=f"Cannot select more than {config.max_selection} items",
145 errors={ # type: ignore[arg-type]
146 "selection": [
147 f"Maximum {config.max_selection} items allowed",
148 ],
149 },
150 ),
151 )
153 # Custom validation
154 if hasattr(handler, "validate"):
155 validation_result = await handler.validate(context, config)
156 if validation_result.is_err():
157 return validation_result
159 # Execute based on mode
160 try:
161 if config.execution_mode == ActionExecutionMode.ASYNC:
162 return await self._execute(context, config, handler) # type: ignore[return-value]
163 result: Result[ActionResult, AdminError] = await self._execute_direct(
164 context, handler
165 )
166 await self._publish_action_notification(context, config, result)
167 return result # type: ignore[return-value]
168 except (RuntimeError, ValueError, TypeError, OSError) as e:
169 if hasattr(handler, "on_failure"):
170 await handler.on_failure(context, e)
171 await self._publish_action_failure(
172 context, config.label or context.action_name, str(e)
173 )
174 return Err(
175 AdminError(
176 message=str(e),
177 ),
178 )
180 async def _execute_direct(
181 self,
182 context: ActionContext,
183 handler: ActionHandler,
184 ) -> Result[ActionResult, AdminError]:
185 """Execute action directly (sync mode, but async handler)."""
186 result = await handler.execute(context)
188 if hasattr(handler, "on_success"):
189 await handler.on_success(context, result)
191 return Ok(result)
193 async def _execute(
194 self,
195 context: ActionContext,
196 config: ActionConfig,
197 handler: ActionHandler,
198 ) -> Result[ActionResult, AdminError]:
199 """Execute action asynchronously via task queue."""
200 if not self.task_scheduler:
201 # Fallback to direct execution
202 return await self._execute_direct(context, handler)
204 # Schedule task
205 task_id = await self.task_scheduler.schedule(
206 f"admin.action.{context.resource_name}.{context.action_name}",
207 args=(context,),
208 kwargs={},
209 )
211 return Ok(
212 ActionResult.async_started(
213 task_id=task_id,
214 message=f"Action '{config.label}' has been scheduled",
215 ),
216 )
218 async def _publish_action_notification(
219 self,
220 context: ActionContext,
221 config: ActionConfig,
222 result: Result[ActionResult, AdminError],
223 ) -> None:
224 """Publish a notification event for a completed action."""
225 if not self.event_hub:
226 return
227 if result.is_ok():
228 action_result = result.unwrap()
229 await self.event_hub.publish_notification(
230 title=f"Action completed: {config.label or context.action_name}",
231 message=action_result.message[:200]
232 if action_result.message
233 else f"Action '{context.action_name}' succeeded",
234 level="success",
235 target_users=[getattr(context.user, "id", None)]
236 if context.user
237 else None,
238 )
240 async def _publish_action_failure(
241 self,
242 context: ActionContext,
243 action_label: str,
244 error: str,
245 ) -> None:
246 """Publish a notification event for a failed action."""
247 if not self.event_hub:
248 return
249 await self.event_hub.publish_notification(
250 title=f"Action failed: {action_label}",
251 message=error[:200],
252 level="error",
253 target_users=[getattr(context.user, "id", None)] if context.user else None,
254 )
256 def get_available_actions(
257 self,
258 resource_name: str,
259 user_permissions: set[str] | None = None,
260 context: str = "list", # list, detail
261 ) -> list[ActionConfig]:
262 """Get actions available for a resource.
264 Args:
265 resource_name: Resource name
266 user_permissions: Optional permissions to filter by
267 context: Where actions will be shown (list or detail)
269 Returns:
270 List of available action configurations
271 """
272 actions = self.registry.get_actions_for_resource(resource_name)
274 # Filter by context
275 if context == "list":
276 actions = list(filter(lambda a: a.show_in_list, actions))
277 elif context == "detail":
278 actions = list(filter(lambda a: a.show_in_detail, actions))
280 # Filter by permissions
281 if user_permissions:
282 actions = [
283 a
284 for a in actions
285 if not a.permission or a.permission in user_permissions
286 ]
288 return actions
291# Decorator for registering actions
294def action(
295 name: str,
296 label: str,
297 *,
298 resource: str | None = None,
299 action_type: ActionType = ActionType.SINGLE,
300 execution_mode: ActionExecutionMode = ActionExecutionMode.SYNC,
301 permission: str | None = None,
302 confirm_message: str | None = None,
303 icon: str | None = None,
304 **kwargs: Any,
305) -> Callable[[Callable[[ActionContext], Any]], Callable[[ActionContext], Any]]:
306 """Decorator to register an action handler.
308 Example:
309 @action("deactivate", "Deactivate User", resource="users", confirm_message="Deactivate this user?")
310 async def deactivate_user(context: ActionContext) -> ActionResult:
311 # implementation
312 return ActionResult(message="User deactivated")
313 """
315 def decorator(
316 func: Callable[[ActionContext], Any],
317 ) -> Callable[[ActionContext], Any]:
318 # Store action metadata on the function
319 func._action_config = ActionConfig( # type: ignore[attr-defined]
320 name=name,
321 label=label,
322 action_type=action_type,
323 execution_mode=execution_mode,
324 permission=permission,
325 confirm_message=confirm_message,
326 icon=icon,
327 **kwargs,
328 )
329 func._action_resource = resource # type: ignore[attr-defined]
330 return func
332 return decorator
335__all__ = [
336 "AbstractActionHandler",
337 "ActionConfig",
338 "ActionContext",
339 "ActionExecutionMode",
340 "ActionExecutor",
341 "ActionHandler",
342 "ActionRegistry",
343 "ActionResult",
344 "ActionType",
345 "ActionValidator",
346 "FunctionActionHandler",
347 "action",
348]