Coverage for src/lexigram/admin/services/action_executor.py: 81%
122 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +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.realtime import SubjectAdminEventHub
17from lexigram.admin.services.action_registry import (
18 ActionConfig,
19 ActionContext,
20 ActionExecutionMode,
21 ActionHandler,
22 ActionRegistry,
23 ActionResult,
24 ActionType,
25)
26from lexigram.contracts.admin.action_hooks import ActionHookProtocol
27from lexigram.contracts.auth import AuthorizerProtocol
28from lexigram.di.decorators import inject
29from lexigram.result import Err, Ok, Result
32@runtime_checkable
33class TaskScheduler(Protocol):
34 """Protocol for scheduling background tasks."""
36 async def schedule(
37 self,
38 task_name: str,
39 args: tuple[Any, ...],
40 kwargs: dict[str, Any],
41 ) -> str: ...
44@inject
45class ActionExecutor:
46 """Executes admin actions with validation and authorization.
48 Handles:
49 - Action lookup and validation
50 - Authorization checks
51 - Sync and async execution
52 - Result handling
53 - Real-time notification publishing via SubjectAdminEventHub
55 Example:
56 >>> executor = ActionExecutor(registry, authorizer)
57 >>> context = ActionContext(
58 ... user=current_user,
59 ... resource_name="users",
60 ... action_name="deactivate",
61 ... record_id=123,
62 ... )
63 >>> result = await executor.execute(context)
64 """
66 def __init__(
67 self,
68 registry: ActionRegistry,
69 authorizer: AuthorizerProtocol | None = None,
70 task_scheduler: TaskScheduler | None = None,
71 event_hub: SubjectAdminEventHub | None = None,
72 resource_resolver: Any | None = None,
73 ):
74 """Initialize the action executor.
76 Args:
77 registry: Action registry
78 authorizer: Optional authorizer for permission checks
79 task_scheduler: Optional task scheduler for async actions
80 event_hub: Optional SubjectAdminEventHub for publishing
81 real-time notifications. Uses on_overflow="drop_latest"
82 internally, so publishing here never blocks this
83 method's caller on a slow subscriber.
84 resource_resolver: Optional callable ``(resource_name) -> Resource``
85 used to resolve resource-level action hooks via
86 ``Resource.get_action_hooks(action_name)``.
87 """
88 self.registry = registry
89 self.authorizer = authorizer
90 self.task_scheduler = task_scheduler
91 self.event_hub = event_hub
92 self.resource_resolver = resource_resolver
94 async def execute(
95 self,
96 context: ActionContext,
97 ) -> Result[
98 ActionResult, PermissionDeniedError | AdminValidationError | AdminError
99 ]:
100 """Execute an action.
102 Args:
103 context: Action execution context
105 Returns:
106 Result containing ActionResult or error
107 """
108 # Get action
109 action = self.registry.get(context.resource_name, context.action_name)
110 if not action:
111 action = self.registry.get_global(context.action_name)
113 if not action:
114 return Err(
115 AdminError(
116 message=f"Action '{context.action_name}' not found for resource '{context.resource_name}'",
117 ),
118 )
120 config, handler = action
122 # Authorization
123 if self.authorizer and config.permission:
124 can_execute = await self.authorizer.can_execute_action(
125 context.user,
126 context.resource_name,
127 context.action_name,
128 )
129 if not can_execute:
130 return Err(
131 PermissionDeniedError(
132 resource=context.resource_name,
133 action=context.action_name,
134 message=f"No permission to execute '{context.action_name}'",
135 ),
136 )
138 # Validate bulk constraints
139 if config.action_type == ActionType.BULK:
140 if context.target_count < config.min_selection:
141 return Err(
142 AdminValidationError(
143 message=f"Select at least {config.min_selection} item(s)",
144 errors={ # type: ignore[arg-type]
145 "selection": [
146 f"Minimum {config.min_selection} items required",
147 ],
148 },
149 ),
150 )
151 if config.max_selection and context.target_count > config.max_selection:
152 return Err(
153 AdminValidationError(
154 message=f"Cannot select more than {config.max_selection} items",
155 errors={ # type: ignore[arg-type]
156 "selection": [
157 f"Maximum {config.max_selection} items allowed",
158 ],
159 },
160 ),
161 )
163 # Custom validation
164 if hasattr(handler, "validate"):
165 validation_result = await handler.validate(context, config)
166 if validation_result.is_err():
167 return validation_result
169 # Run before hooks (may amend data or abort the action)
170 hooks_result = await self._run_before_hooks(context, handler)
171 if hooks_result.is_err():
172 error = hooks_result.unwrap_err()
173 await self._run_failure_hooks(context, handler, error)
174 await self._publish_action_failure(
175 context, config.label or context.action_name, str(error)
176 )
177 return Err(
178 AdminError(
179 message=getattr(error, "message", None) or str(error),
180 ),
181 )
183 # Execute based on mode
184 try:
185 if config.execution_mode == ActionExecutionMode.ASYNC:
186 return await self._execute(context, config, handler) # type: ignore[return-value]
187 result: Result[ActionResult, AdminError] = await self._execute_direct(
188 context, handler
189 )
190 await self._publish_action_notification(context, config, result)
191 return result # type: ignore[return-value]
192 except (RuntimeError, ValueError, TypeError, OSError) as e:
193 if hasattr(handler, "on_failure"):
194 await handler.on_failure(context, e)
195 await self._run_failure_hooks(context, handler, e)
196 await self._publish_action_failure(
197 context, config.label or context.action_name, str(e)
198 )
199 return Err(
200 AdminError(
201 message=str(e),
202 ),
203 )
205 async def _execute_direct(
206 self,
207 context: ActionContext,
208 handler: ActionHandler,
209 ) -> Result[ActionResult, AdminError]:
210 """Execute action directly (sync mode, but async handler)."""
211 result = await handler.execute(context)
213 if hasattr(handler, "on_success"):
214 await handler.on_success(context, result)
216 await self._run_after_hooks(context, handler, result)
218 return Ok(result)
220 async def _run_before_hooks(
221 self,
222 context: ActionContext,
223 handler: ActionHandler,
224 ) -> Result[None, Exception]:
225 """Run before hooks for an action.
227 Collects hooks from the handler (``HasActionHooks``) and from the
228 resource via ``Resource.get_action_hooks``. Each hook may amend
229 ``context.parameters``; returning ``Err`` aborts the action.
231 Args:
232 context: Action execution context
233 handler: Action handler
235 Returns:
236 ``Ok(None)`` if all hooks passed, ``Err`` to abort the action.
237 """
238 for hook in self._collect_hooks(context, handler, "before"):
239 result = await hook.before(context, context.parameters)
240 if result.is_err():
241 return Err(result.unwrap_err())
242 amended = result.unwrap()
243 if amended:
244 context.parameters.update(amended)
245 return Ok(None)
247 async def _run_after_hooks(
248 self,
249 context: ActionContext,
250 handler: ActionHandler,
251 action_result: ActionResult,
252 ) -> None:
253 """Run after hooks for a successful action execution.
255 Args:
256 context: Action execution context
257 handler: Action handler
258 action_result: Result of the action body
259 """
260 for hook in self._collect_hooks(context, handler, "after"):
261 await hook.after(context, action_result)
263 async def _run_failure_hooks(
264 self,
265 context: ActionContext,
266 handler: ActionHandler,
267 error: Exception,
268 ) -> None:
269 """Run failure hooks when an action fails.
271 Args:
272 context: Action execution context
273 handler: Action handler
274 error: The error that caused the failure
275 """
276 for hook in self._collect_hooks(context, handler, "failure"):
277 await hook.on_failure(context, error)
279 def _collect_hooks(
280 self, context: ActionContext, handler: ActionHandler, stage: str
281 ) -> list[ActionHookProtocol]:
282 """Collect lifecycle hooks for an action.
284 Hooks come from two sources:
285 - Handler-level: ``before_hooks`` / ``after_hooks`` / ``failure_hooks``
286 attributes on the handler (``HasActionHooks``).
287 - Resource-level: ``Resource.get_action_hooks(action_name)`` resolved
288 via ``resource_resolver``.
290 Args:
291 context: Action execution context
292 handler: Action handler
293 stage: One of ``"before"``, ``"after"``, ``"failure"``
295 Returns:
296 List of hooks to run for the stage.
297 """
298 hooks: list[ActionHookProtocol] = []
299 handler_hooks = getattr(handler, f"{stage}_hooks", None)
300 if handler_hooks:
301 hooks.extend(handler_hooks)
302 if self.resource_resolver:
303 resource = self.resource_resolver(context.resource_name)
304 if resource is not None:
305 resource_hooks = getattr(resource, "get_action_hooks", None)
306 if resource_hooks:
307 hooks.extend(resource_hooks(context.action_name))
308 return hooks
310 async def _execute(
311 self,
312 context: ActionContext,
313 config: ActionConfig,
314 handler: ActionHandler,
315 ) -> Result[ActionResult, AdminError]:
316 """Execute action asynchronously via task queue."""
317 if not self.task_scheduler:
318 # Fallback to direct execution
319 return await self._execute_direct(context, handler)
321 # Schedule task
322 task_id = await self.task_scheduler.schedule(
323 f"admin.action.{context.resource_name}.{context.action_name}",
324 args=(context,),
325 kwargs={},
326 )
328 return Ok(
329 ActionResult.async_started(
330 task_id=task_id,
331 message=f"Action '{config.label}' has been scheduled",
332 ),
333 )
335 async def _publish_action_notification(
336 self,
337 context: ActionContext,
338 config: ActionConfig,
339 result: Result[ActionResult, AdminError],
340 ) -> None:
341 """Publish a notification event for a completed action."""
342 if not self.event_hub:
343 return
344 if result.is_ok():
345 action_result = result.unwrap()
346 await self.event_hub.publish_notification(
347 title=f"Action completed: {config.label or context.action_name}",
348 message=action_result.message[:200]
349 if action_result.message
350 else f"Action '{context.action_name}' succeeded",
351 level="success",
352 target_users=[getattr(context.user, "id", None)]
353 if context.user
354 else None,
355 )
357 async def _publish_action_failure(
358 self,
359 context: ActionContext,
360 action_label: str,
361 error: str,
362 ) -> None:
363 """Publish a notification event for a failed action."""
364 if not self.event_hub:
365 return
366 await self.event_hub.publish_notification(
367 title=f"Action failed: {action_label}",
368 message=error[:200],
369 level="error",
370 target_users=[getattr(context.user, "id", None)] if context.user else None,
371 )
373 def get_available_actions(
374 self,
375 resource_name: str,
376 user_permissions: set[str] | None = None,
377 context: str = "list", # list, detail
378 ) -> list[ActionConfig]:
379 """Get actions available for a resource.
381 Args:
382 resource_name: Resource name
383 user_permissions: Optional permissions to filter by
384 context: Where actions will be shown (list or detail)
386 Returns:
387 List of available action configurations
388 """
389 actions = self.registry.get_actions_for_resource(resource_name)
391 # Filter by context
392 if context == "list":
393 actions = list(filter(lambda a: a.show_in_list, actions))
394 elif context == "detail":
395 actions = list(filter(lambda a: a.show_in_detail, actions))
397 # Filter by permissions
398 if user_permissions:
399 actions = [
400 a
401 for a in actions
402 if not a.permission or a.permission in user_permissions
403 ]
405 return actions
408# Decorator for registering actions
411def action(
412 name: str,
413 label: str,
414 *,
415 resource: str | None = None,
416 action_type: ActionType = ActionType.SINGLE,
417 execution_mode: ActionExecutionMode = ActionExecutionMode.SYNC,
418 permission: str | None = None,
419 confirm_message: str | None = None,
420 icon: str | None = None,
421 **kwargs: Any,
422) -> Callable[[Callable[[ActionContext], Any]], Callable[[ActionContext], Any]]:
423 """Decorator to register an action handler.
425 Example:
426 @action("deactivate", "Deactivate User", resource="users", confirm_message="Deactivate this user?")
427 async def deactivate_user(context: ActionContext) -> ActionResult:
428 # implementation
429 return ActionResult(message="User deactivated")
430 """
432 def decorator(
433 func: Callable[[ActionContext], Any],
434 ) -> Callable[[ActionContext], Any]:
435 # Store action metadata on the function
436 func._action_config = ActionConfig( # type: ignore[attr-defined]
437 name=name,
438 label=label,
439 action_type=action_type,
440 execution_mode=execution_mode,
441 permission=permission,
442 confirm_message=confirm_message,
443 icon=icon,
444 **kwargs,
445 )
446 func._action_resource = resource # type: ignore[attr-defined]
447 return func
449 return decorator
452__all__ = [
453 "AbstractActionHandler",
454 "ActionConfig",
455 "ActionContext",
456 "ActionExecutionMode",
457 "ActionExecutor",
458 "ActionHandler",
459 "ActionRegistry",
460 "ActionResult",
461 "ActionType",
462 "ActionValidator",
463 "FunctionActionHandler",
464 "action",
465]