Coverage for src/lexigram/admin/resources/action_handlers.py: 0%
211 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""Record-level CRUD action handlers for Admin Resources.
3List / detail / create / edit / clone / restore / purge / delete flows,
4plus the form-data coercion helpers they share. Specialized handlers
5(import, user permissions, bulk) live in :mod:`.handler`.
6"""
8from __future__ import annotations
10from typing import Any, Protocol
12from starlette.requests import Request as StarletteRequest
13from starlette.responses import HTMLResponse
15from lexigram.admin.resources.form_coercion import (
16 _validation_errors_to_dict,
17)
18from lexigram.logging import get_logger
20logger = get_logger(__name__)
23class ResourceActionHandler(Protocol):
24 """Protocol for action handlers."""
26 def can_handle(self, action: str) -> bool: ...
28 async def handle(
29 self, request: StarletteRequest, resource: Any, **kwargs: Any
30 ) -> Any: ...
33class ListActionHandler:
34 def __init__(self, list_renderer: Any):
35 self.list_renderer = list_renderer
37 def can_handle(self, action: str) -> bool:
38 return action == "list"
40 async def handle(
41 self, request: StarletteRequest, resource: Any, **kwargs: Any
42 ) -> Any:
43 return await self.list_renderer.render(request, resource)
46class DetailActionHandler:
47 def __init__(self, detail_renderer: Any):
48 self.detail_renderer = detail_renderer
50 def can_handle(self, action: str) -> bool:
51 return action == "detail"
53 async def handle(
54 self, request: StarletteRequest, resource: Any, **kwargs: Any
55 ) -> Any:
56 item_id = request.path_params.get("id", "?")
57 return await self.detail_renderer.render_detail(request, resource, item_id)
60class CreateActionHandler:
61 def __init__(self, form_renderer: Any):
62 self.form_renderer = form_renderer
64 def can_handle(self, action: str) -> bool:
65 return action == "create"
67 async def handle(
68 self, request: StarletteRequest, resource: Any, **kwargs: Any
69 ) -> Any:
70 if request.method == "POST":
71 return await self._handle_create(request, resource)
72 return await self.form_renderer.render_create(request, resource)
74 async def _handle_create(self, request: StarletteRequest, resource: Any) -> Any:
75 from lexigram.admin.resources.base import Resource
77 form = request.scope.get("admin_form_data") or await request.form()
78 data = dict(form)
79 data.pop("csrf_token", None)
81 if isinstance(resource, Resource) and resource._data_source:
82 validation = await resource.before_validate(data)
83 if validation.is_err():
84 error = validation.unwrap_err()
85 return await self.form_renderer.render_create(
86 request, resource, errors=_validation_errors_to_dict(error)
87 )
89 validated_data = validation.unwrap()
90 validated = await resource.before_create(validated_data)
91 record = await resource._data_source.create(validated)
92 await resource.after_create(record)
94 resource_prefix = request.scope.get(
95 "admin_resource_prefix", resource.name or ""
96 )
98 from starlette.responses import HTMLResponse
100 return HTMLResponse(
101 f'<html><head><meta http-equiv="refresh" content="0;url=/admin/{resource_prefix}"></head><body></body></html>'
102 )
104 return await self.form_renderer.render_create(request, resource)
107class EditActionHandler:
108 def __init__(self, form_renderer: Any):
109 self.form_renderer = form_renderer
111 def can_handle(self, action: str) -> bool:
112 return action == "edit"
114 async def handle(
115 self, request: StarletteRequest, resource: Any, **kwargs: Any
116 ) -> Any:
117 item_id = request.path_params.get("id", "?")
118 if request.method == "POST":
119 return await self._handle_update(request, resource, item_id)
120 return await self.form_renderer.render_edit(request, resource, item_id)
122 async def _handle_update(
123 self, request: StarletteRequest, resource: Any, item_id: str
124 ) -> Any:
125 from lexigram.admin.resources.base import Resource
127 form = request.scope.get("admin_form_data") or await request.form()
128 data = dict(form)
129 data.pop("csrf_token", None)
131 if isinstance(resource, Resource) and resource._data_source:
132 validation = await resource.before_validate(data)
133 if validation.is_err():
134 error = validation.unwrap_err()
135 return await self.form_renderer.render_edit(
136 request,
137 resource,
138 item_id,
139 errors=_validation_errors_to_dict(error),
140 )
142 validated_data = validation.unwrap()
143 validated = await resource.before_update(item_id, validated_data)
144 record = await resource._data_source.find_one(item_id)
145 can_update = getattr(resource, "can_update", None)
146 if can_update and not can_update(record):
147 return HTMLResponse("This record cannot be updated", status_code=403)
148 updated_record = await resource._data_source.update(item_id, validated)
149 await resource.after_update(updated_record)
151 resource_prefix = request.scope.get(
152 "admin_resource_prefix", resource.name or ""
153 )
155 return HTMLResponse(
156 f'<html><head><meta http-equiv="refresh" content="0;url=/admin/{resource_prefix}"></head><body></body></html>'
157 )
159 return await self.form_renderer.render_edit(request, resource, item_id)
162class CloneActionHandler:
163 """Handler for the ``clone`` action — duplicates a record and redirects."""
165 def can_handle(self, action: str) -> bool:
166 return action == "clone"
168 async def handle(
169 self, request: StarletteRequest, resource: Any, **kwargs: Any
170 ) -> Any:
171 from lexigram.admin.resources.base import Resource
173 item_id = request.path_params.get("id", "?")
175 if not isinstance(resource, Resource):
176 return HTMLResponse(
177 "<h1>Clone not supported for this resource</h1>", status_code=400
178 )
180 new_record = await resource.duplicate(item_id)
181 new_id = str(getattr(new_record, "id", "?"))
182 resource_prefix = request.scope.get(
183 "admin_resource_prefix", resource.name or ""
184 )
185 from starlette.responses import RedirectResponse
187 return RedirectResponse(
188 url=f"/admin/{resource_prefix}/{new_id}/edit",
189 status_code=302,
190 )
193class RestoreActionHandler:
194 """Handler for the ``restore`` action — restores a soft-deleted record and redirects."""
196 def can_handle(self, action: str) -> bool:
197 return action == "restore"
199 async def handle(
200 self, request: StarletteRequest, resource: Any, **kwargs: Any
201 ) -> Any:
202 from lexigram.admin.resources.base import Resource
204 item_id = request.path_params.get("id", "?")
206 if not isinstance(resource, Resource):
207 return HTMLResponse(
208 "<h1>Restore not supported for this resource</h1>", status_code=400
209 )
211 restored = await resource.restore(item_id)
212 new_id = str(getattr(restored, "id", "?"))
213 resource_prefix = request.scope.get(
214 "admin_resource_prefix", resource.name or ""
215 )
216 from starlette.responses import RedirectResponse
218 return RedirectResponse(
219 url=f"/admin/{resource_prefix}/{new_id}/edit",
220 status_code=302,
221 )
224class PurgeActionHandler:
225 """Handler for the ``purge`` action — permanently deletes a record and redirects."""
227 def can_handle(self, action: str) -> bool:
228 return action == "purge"
230 async def handle(
231 self, request: StarletteRequest, resource: Any, **kwargs: Any
232 ) -> Any:
233 from lexigram.admin.resources.base import Resource
235 item_id = request.path_params.get("id", "?")
237 if not isinstance(resource, Resource):
238 return HTMLResponse(
239 "<h1>Purge not supported for this resource</h1>", status_code=400
240 )
242 await resource.purge(item_id)
243 resource_prefix = request.scope.get(
244 "admin_resource_prefix", resource.name or ""
245 )
246 from starlette.responses import RedirectResponse
248 return RedirectResponse(
249 url=f"/admin/{resource_prefix}",
250 status_code=302,
251 )
254class ImportActionHandler:
255 """Handler for import download routes (example CSV, failed-import report).
257 Serves GET ``import-example`` (the resource's declared
258 :class:`~lexigram.admin.actions.standard.ImportAction` template) and
259 GET ``import-report`` (a stored failed-import report as CSV).
260 """
262 _ACTIONS = ("import-example", "import-report")
264 def can_handle(self, action: str) -> bool:
265 """Whether this handler serves the given route action."""
266 return action in self._ACTIONS
268 @staticmethod
269 def _find_import_action(resource: Any) -> Any:
270 """Locate the resource's declared ImportAction, if any."""
271 from lexigram.admin.actions.standard import ImportAction
273 for collection in ("header_actions", "actions"):
274 for action in getattr(resource, collection, None) or []:
275 if isinstance(action, ImportAction):
276 return action
277 return None
279 @staticmethod
280 def _csv_response(content: str, filename: str) -> Any:
281 """Build an attachment CSV response."""
282 from starlette.responses import Response
284 return Response(
285 content=content,
286 media_type="text/csv",
287 headers={"Content-Disposition": f'attachment; filename="{filename}"'},
288 )
290 async def handle(
291 self, request: StarletteRequest, resource: Any, **kwargs: Any
292 ) -> Any:
293 from starlette.responses import HTMLResponse
295 action = self._find_import_action(resource)
296 if action is None:
297 return HTMLResponse(
298 "<h1>Import not configured for this resource</h1>",
299 status_code=404,
300 )
301 if request.method != "GET":
302 return HTMLResponse("Method not allowed", status_code=405)
304 requested = request.scope.get("admin_action", "")
305 if requested == "import-example":
306 content = action.example_csv()
307 if not content:
308 return HTMLResponse(
309 "<h1>No example CSV configured</h1>", status_code=404
310 )
311 return self._csv_response(content, action.example_filename)
313 report_id = request.query_params.get("report_id", "")
314 content = action.report_csv(report_id)
315 if content is None:
316 return HTMLResponse("<h1>Report not found</h1>", status_code=404)
317 filename = action.report_filename(report_id) or "import-errors.csv"
318 return self._csv_response(content, filename)
321class DeleteActionHandler:
322 """Handler for delete-confirm and delete actions."""
324 def can_handle(self, action: str) -> bool:
325 return action in ("delete-confirm", "delete")
327 async def handle(
328 self, request: StarletteRequest, resource: Any, **kwargs: Any
329 ) -> Any:
330 item_id = request.path_params.get("id", "?")
332 if request.method == "GET":
333 return await self._confirm_delete(request, resource, item_id)
334 return await self._execute_delete(request, resource, item_id)
336 async def _confirm_delete(
337 self, request: StarletteRequest, resource: Any, item_id: str
338 ) -> Any:
339 from lexigram.admin.resources.base import Resource as AdminResource
341 label = getattr(resource, "label", "Record")
342 record_label = f"{label} #{item_id}"
344 if isinstance(resource, AdminResource) and resource._data_source:
345 try:
346 item = await resource._data_source.find_one(item_id)
347 if item:
348 for field in ("name", "title", "email", "username", "label"):
349 val = (
350 item.get(field)
351 if isinstance(item, dict)
352 else getattr(item, field, None)
353 )
354 if val:
355 record_label = str(val)
356 break
357 except Exception: # noqa: S110 — intentional best-effort fallback
358 pass
360 resource_prefix = request.scope.get("admin_resource_prefix", "")
361 delete_url = f"/admin/{resource_prefix}/{item_id}/delete"
363 from lexigram.admin.ui.organisms.admin_slide_over import render_delete_confirm
365 html = render_delete_confirm(
366 record_label=record_label,
367 delete_url=delete_url,
368 )
369 return HTMLResponse(html)
371 async def _execute_delete(
372 self, request: StarletteRequest, resource: Any, item_id: str
373 ) -> Any:
374 from lexigram.admin.resources.base import Resource as AdminResource
376 if isinstance(resource, AdminResource) and resource._data_source:
377 item = await resource._data_source.find_one(item_id)
378 if item is None:
379 return HTMLResponse("Not found", status_code=404)
381 can_delete = getattr(resource, "can_delete", None)
382 if can_delete and not can_delete(item):
383 is_htmx = request.headers.get("HX-Request") == "true"
384 if is_htmx:
385 response = HTMLResponse("")
386 response.headers["HX-Trigger"] = (
387 '{"show-toast":{"message":"This record cannot be deleted","type":"error"}}'
388 )
389 return response
390 return HTMLResponse(
391 '<html><head><meta http-equiv="refresh" content="0;url=/admin/"></head>'
392 "<body>This record cannot be deleted</body></html>",
393 status_code=409,
394 )
396 success = await resource._data_source.delete(item_id)
397 if not success:
398 return HTMLResponse("Not found", status_code=404)
400 after_delete = getattr(resource, "after_delete", None)
401 if after_delete:
402 await after_delete(item_id)
404 is_htmx = request.headers.get("HX-Request") == "true"
405 resource_prefix = request.scope.get("admin_resource_prefix", "")
407 if is_htmx:
408 response = HTMLResponse("")
409 response.headers["HX-Trigger"] = (
410 '{"refresh-list":true,"show-toast":{"message":"Deleted successfully","type":"success"}}'
411 )
412 response.headers["HX-Redirect"] = f"/admin/{resource_prefix}"
413 return response
415 return HTMLResponse(
416 f'<html><head><meta http-equiv="refresh" content="0;url=/admin/{resource_prefix}"></head><body></body></html>'
417 )
419 return HTMLResponse("Delete not supported", status_code=400)