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