Coverage for src/lexigram/admin/actions/standard/row.py: 32%
153 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:39 +0800
1"""Ready-to-use row actions (edit, view, delete, clone, restore, purge).
3Part of the ``lexigram.admin.actions.standard`` package.
4"""
6from __future__ import annotations
8from collections.abc import Awaitable, Callable
10import inspect
11from typing import Any
13from lexigram.admin.actions.base import RowAction
14from lexigram.admin.actions.standard.utils import _extract_id, _resolve_data_source
15from lexigram.admin.actions.exceptions import ActionError
16from lexigram.admin.actions.types import (
17 ActionColor,
18 ActionContext,
19 ConfirmationConfig,
20)
21from lexigram.ui import Zones
22from lexigram.result import Err, Ok, Result
25class EditAction(RowAction):
26 """Edit a single record."""
28 def __init__(
29 self,
30 name: str = "edit",
31 label: str | None = None,
32 ) -> None:
33 super().__init__(
34 name=name,
35 label=label or "Edit",
36 icon="pencil",
37 color=ActionColor.PRIMARY,
38 )
40 def _get_url(self, record: Any, ctx: ActionContext) -> str | None:
41 record_id = self._get_record_id(record)
42 if not record_id:
43 return None
44 prefix = ctx.resource_prefix or f"/{ctx.resource_name}"
45 return f"{prefix}/{record_id}/edit"
47 def _get_htmx_attrs(
48 self, url: str, record: Any, ctx: ActionContext
49 ) -> dict[str, str]:
50 return {
51 "hx-get": url,
52 "hx-target": Zones.SLIDE_OVER.selector,
53 "hx-swap": Zones.SLIDE_OVER.swap_mode.value,
54 }
56 async def execute(self, record: Any, ctx: ActionContext) -> Result[Any, Any]:
57 return Ok({"message": f"Edited {record}"})
60class ViewAction(RowAction):
61 """View a single record."""
63 def __init__(
64 self,
65 name: str = "view",
66 label: str | None = None,
67 ) -> None:
68 super().__init__(
69 name=name,
70 label=label or "View",
71 icon="eye",
72 color=ActionColor.GRAY,
73 )
75 def _get_url(self, record: Any, ctx: ActionContext) -> str | None:
76 record_id = self._get_record_id(record)
77 if not record_id:
78 return None
79 prefix = ctx.resource_prefix or f"/{ctx.resource_name}"
80 return f"{prefix}/{record_id}"
82 async def execute(self, record: Any, ctx: ActionContext) -> Result[Any, Any]:
83 return Ok({"message": f"Viewed {record}"})
86class DeleteAction(RowAction):
87 """Delete a single record with confirmation."""
89 def __init__(
90 self,
91 name: str = "delete",
92 label: str | None = None,
93 confirm_title: str = "Delete Record",
94 confirm_message: str | None = None,
95 ) -> None:
96 super().__init__(
97 name=name,
98 label=label or "Delete",
99 icon="trash",
100 color=ActionColor.DANGER,
101 )
102 self._confirm_title = confirm_title
103 self._confirm_message = confirm_message or (
104 "Are you sure you want to delete this record? This action cannot be undone."
105 )
107 def _get_url(self, record: Any, ctx: ActionContext) -> str | None:
108 record_id = self._get_record_id(record)
109 if not record_id:
110 return None
111 prefix = ctx.resource_prefix or f"/{ctx.resource_name}"
112 return f"{prefix}/{record_id}/delete-confirm"
114 def _get_htmx_attrs(
115 self, url: str, record: Any, ctx: ActionContext
116 ) -> dict[str, str]:
117 return {
118 "hx-get": url,
119 "hx-target": Zones.SLIDE_OVER.selector,
120 "hx-swap": Zones.SLIDE_OVER.swap_mode.value,
121 "hx-push-url": "false",
122 }
124 def confirm(self) -> ConfirmationConfig | None:
125 return ConfirmationConfig(
126 title=self._confirm_title,
127 message=self._confirm_message,
128 style=ActionColor.DANGER,
129 )
131 async def execute(self, record: Any, ctx: ActionContext) -> Result[Any, Any]:
132 return Ok({"message": f"Deleted {record}", "deleted": True})
135class PermissionsAction(RowAction):
136 """Edit a user's direct permissions (users resource only)."""
138 def __init__(
139 self,
140 name: str = "permissions",
141 label: str | None = None,
142 ) -> None:
143 super().__init__(
144 name=name,
145 label=label or "Edit Permissions",
146 icon="shield",
147 color=ActionColor.GRAY,
148 )
150 def _get_url(self, record: Any, ctx: ActionContext) -> str | None:
151 record_id = self._get_record_id(record)
152 if not record_id:
153 return None
154 prefix = ctx.resource_prefix or f"/{ctx.resource_name}"
155 return f"{prefix}/{record_id}/permissions"
157 async def execute(self, record: Any, ctx: ActionContext) -> Result[Any, Any]:
158 return Ok({"message": f"Editing permissions for {record}"})
161class ImpersonateAction(RowAction):
162 """Impersonate a user's session (users resource only).
164 Target-role restriction (denying impersonation of another
165 super-admin) is enforced server-side only — this action has no
166 access to DI/config at render time (``Action`` is a frozen,
167 import-time-constructed dataclass), and the row's rendered fields
168 don't expose RBAC role membership. See
169 ``docs/superpowers/specs/2026-08-19-admin-impersonation-usability-design.md``
170 D5 for the full reasoning.
171 """
173 def __init__(
174 self,
175 name: str = "impersonate",
176 label: str | None = None,
177 ) -> None:
178 super().__init__(
179 name=name,
180 label=label or "Impersonate",
181 icon="user-check",
182 color=ActionColor.GRAY,
183 )
185 def visible_for(self, record: Any, user: Any | None = None) -> bool:
186 if user is None:
187 return True
188 record_id = self._get_record_id(record)
189 actor_id = str(getattr(user, "id", ""))
190 return record_id != actor_id
192 def _get_url(self, record: Any, ctx: ActionContext) -> str | None:
193 record_id = self._get_record_id(record)
194 if not record_id:
195 return None
196 return f"/admin/impersonate/{record_id}"
198 def _get_htmx_attrs(
199 self, url: str, record: Any, ctx: ActionContext
200 ) -> dict[str, str]:
201 name = (
202 record.get("name", "this user") if isinstance(record, dict) else "this user"
203 )
204 return {
205 "hx-post": url,
206 "hx-target": "body",
207 "hx-swap": "none",
208 "hx-confirm": f"Impersonate {name}?",
209 }
211 async def execute(self, record: Any, ctx: ActionContext) -> Result[Any, Any]:
212 return Ok({"message": f"Impersonating {record}"})
215class CloneAction(RowAction):
216 """Clone a single record through the data source.
218 Mirrors Filament's ``Replicate`` action: ``exclude_attributes`` drops
219 fields from the copy (``excludeAttributes``), ``mutate_record_data``
220 transforms the record data before creation (``mutateRecordDataUsing``),
221 and ``before_replica_saved`` runs just before the new record is
222 persisted (``beforeReplicaSaved``). The ``id`` field is always stripped
223 so a fresh identifier is assigned.
224 """
226 def __init__(
227 self,
228 name: str = "clone",
229 label: str | None = None,
230 data_source: Any | None = None,
231 exclude_attributes: list[str] | None = None,
232 mutate_record_data: Callable[
233 [dict[str, Any]], dict[str, Any] | Awaitable[dict[str, Any]]
234 ]
235 | None = None,
236 before_replica_saved: Callable[[dict[str, Any]], None | Awaitable[None]]
237 | None = None,
238 ) -> None:
239 super().__init__(
240 name=name,
241 label=label or "Clone",
242 icon="copy",
243 color=ActionColor.SECONDARY,
244 )
245 self._data_source = data_source
246 self._exclude_attributes = exclude_attributes or []
247 self._mutate_record_data = mutate_record_data
248 self._before_replica_saved = before_replica_saved
250 def _get_url(self, record: Any, ctx: ActionContext) -> str | None:
251 record_id = self._get_record_id(record)
252 if not record_id:
253 return None
254 prefix = ctx.resource_prefix or f"/{ctx.resource_name}"
255 return f"{prefix}/{record_id}/clone"
257 async def execute(self, record: Any, ctx: ActionContext) -> Result[Any, Any]:
258 record_id = _extract_id(record)
259 if record_id is None:
260 return Err(ActionError("Clone requires a record with an id."))
261 data_source = _resolve_data_source(ctx, self._data_source)
262 if data_source is None:
263 return Err(
264 ActionError(
265 "Clone requires a data source; inject one or set ctx.data_source."
266 )
267 )
268 original = await data_source.find_one(record_id)
269 if original is None:
270 return Err(ActionError(f"Record {record_id} not found."))
271 data: dict[str, Any] = dict(original) if isinstance(original, dict) else {}
272 if not data and hasattr(original, "__dict__"):
273 data = dict(vars(original))
274 data.pop("id", None)
275 for attribute in self._exclude_attributes:
276 data.pop(attribute, None)
277 if self._mutate_record_data is not None:
278 mutated = self._mutate_record_data(data)
279 data = await mutated if inspect.isawaitable(mutated) else mutated
280 if self._before_replica_saved is not None:
281 hook = self._before_replica_saved(data)
282 if inspect.isawaitable(hook):
283 await hook
284 created = await data_source.create(data)
285 if created is None:
286 return Err(ActionError(f"Failed to clone record {record_id}."))
287 return Ok(
288 {
289 "message": "Cloned record",
290 "record": created,
291 "cloned_id": _extract_id(created),
292 }
293 )
296class RestoreAction(RowAction):
297 """Restore a single soft-deleted record."""
299 def __init__(
300 self,
301 name: str = "restore",
302 label: str | None = None,
303 ) -> None:
304 super().__init__(
305 name=name,
306 label=label or "Restore",
307 icon="rotate-ccw",
308 color=ActionColor.SUCCESS,
309 )
311 def _get_url(self, record: Any, ctx: ActionContext) -> str | None:
312 record_id = self._get_record_id(record)
313 if not record_id:
314 return None
315 prefix = ctx.resource_prefix or f"/{ctx.resource_name}"
316 return f"{prefix}/{record_id}/restore"
318 async def execute(self, record: Any, ctx: ActionContext) -> Result[Any, Any]:
319 return Ok({"message": f"Restored {record}"})
322class PurgeAction(RowAction):
323 """Permanently delete a soft-deleted record."""
325 def __init__(
326 self,
327 name: str = "purge",
328 label: str | None = None,
329 confirm_title: str = "Permanently Delete Record",
330 confirm_message: str | None = None,
331 ) -> None:
332 super().__init__(
333 name=name,
334 label=label or "Purge",
335 icon="trash-2",
336 color=ActionColor.DANGER,
337 )
338 self._confirm_title = confirm_title
339 self._confirm_message = confirm_message or (
340 "Are you sure you want to permanently delete this record? "
341 "This action cannot be undone."
342 )
344 def _get_url(self, record: Any, ctx: ActionContext) -> str | None:
345 record_id = self._get_record_id(record)
346 if not record_id:
347 return None
348 prefix = ctx.resource_prefix or f"/{ctx.resource_name}"
349 return f"{prefix}/{record_id}/purge"
351 def _get_htmx_attrs(
352 self, url: str, record: Any, ctx: ActionContext
353 ) -> dict[str, str]:
354 attrs: dict[str, str] = {
355 "hx-delete": url,
356 "hx-target": "#table-data",
357 "hx-swap": "innerHTML",
358 }
359 confirmation = self.confirm()
360 if confirmation and confirmation.message:
361 attrs["hx-confirm"] = confirmation.message
362 return attrs
364 def confirm(self) -> ConfirmationConfig | None:
365 return ConfirmationConfig(
366 title=self._confirm_title,
367 message=self._confirm_message,
368 style=ActionColor.DANGER,
369 )
371 async def execute(self, record: Any, ctx: ActionContext) -> Result[Any, Any]:
372 return Ok({"message": f"Purged {record}", "purged": True})