Coverage for src/lexigram/admin/resources/handler.py: 64%

1from __future__ import annotations 

2 

3from datetime import date, datetime 

4from decimal import Decimal 

5from decimal import InvalidOperation as DecimalInvalidOp 

6from enum import Enum 

7import types 

8from typing import Any, Protocol, Union, get_args, get_origin, get_type_hints 

9from uuid import UUID 

10 

11from starlette.requests import Request as StarletteRequest 

12from starlette.responses import HTMLResponse, RedirectResponse 

13 

14from lexigram.admin.config import AdminConfig 

15from lexigram.di.decorators import inject 

16from lexigram.logging import get_logger 

17 

18logger = get_logger(__name__) 

19 

20 

21def _unwrap_optional(tp: type) -> type: 

22 """Unwrap Optional[T] or T | None to T.""" 

23 origin = get_origin(tp) 

24 if origin in (Union, types.UnionType): 

25 non_none = [a for a in get_args(tp) if a is not type(None)] 

26 if len(non_none) == 1: 

27 return non_none[0] 

28 return tp 

29 

30 

31def _coerce_form_data(data: dict, model: type | None) -> dict: 

32 """Convert HTML form string values to proper Python types.""" 

33 if model is None: 

34 return data 

35 try: 

36 hints = get_type_hints(model) 

37 except Exception: 

38 return data 

39 

40 for key, value in list(data.items()): 

41 if key not in hints: 

42 continue 

43 if not isinstance(value, str): 

44 continue 

45 

46 expected = _unwrap_optional(hints[key]) 

47 origin = get_origin(expected) 

48 if expected is bool: 

49 data[key] = value == "on" 

50 elif expected is int: 

51 try: 

52 data[key] = int(value) 

53 except (ValueError, TypeError): 

54 pass 

55 elif expected is float: 

56 if value == "": 

57 data[key] = None 

58 else: 

59 try: 

60 data[key] = float(value) 

61 except (ValueError, TypeError): 

62 pass 

63 elif expected is Decimal: 

64 if value == "": 

65 data[key] = None 

66 else: 

67 try: 

68 data[key] = Decimal(value) 

69 except (DecimalInvalidOp, ValueError, TypeError): 

70 pass 

71 elif expected is UUID: 

72 if value == "": 

73 data[key] = None 

74 else: 

75 try: 

76 data[key] = UUID(value) 

77 except (ValueError, AttributeError): 

78 pass 

79 elif expected is datetime: 

80 if value == "": 

81 data[key] = None 

82 else: 

83 try: 

84 data[key] = datetime.fromisoformat(value.replace("Z", "+00:00")) 

85 except (ValueError, TypeError): 

86 pass 

87 elif expected is date: 

88 if value == "": 

89 data[key] = None 

90 else: 

91 try: 

92 data[key] = date.fromisoformat(value) 

93 except (ValueError, TypeError): 

94 pass 

95 elif origin is list: 

96 if value == "": 

97 data[key] = [] 

98 else: 

99 args = get_args(expected) 

100 inner = args[0] if args else str 

101 items = [s.strip() for s in value.split(",")] 

102 if inner is not str: 

103 try: 

104 data[key] = [inner(item) for item in items] 

105 except (ValueError, TypeError): 

106 pass 

107 else: 

108 data[key] = items 

109 elif isinstance(expected, type) and issubclass(expected, Enum): 

110 if value != "": 

111 try: 

112 data[key] = expected(value) 

113 except (ValueError, TypeError): 

114 pass 

115 elif origin is dict: 

116 if value != "": 

117 try: 

118 from lexigram.serialization import loads_str as _json_loads 

119 

120 data[key] = _json_loads(value) 

121 except Exception: # noqa: S110 — intentional best-effort fallback 

122 pass 

123 

124 return data 

125 

126 

127def _validation_errors_to_dict(error: Any) -> dict[str, list[str]]: 

128 """Convert AdminValidationError.errors (list[FieldError]) to dict form.""" 

129 errors: dict[str, list[str]] = {} 

130 for fe in error.errors: 

131 errors.setdefault(fe.field, []).append(fe.message) 

132 return errors 

133 

134 

135class ResourceActionHandler(Protocol): 

136 """Protocol for action handlers.""" 

137 

138 def can_handle(self, action: str) -> bool: ... 

139 

140 async def handle( 

141 self, request: StarletteRequest, resource: Any, **kwargs 

142 ) -> Any: ... 

143 

144 

145class ListActionHandler: 

146 def __init__(self, list_renderer: Any): 

147 self.list_renderer = list_renderer 

148 

149 def can_handle(self, action: str) -> bool: 

150 return action == "list" 

151 

152 async def handle(self, request: StarletteRequest, resource: Any, **kwargs) -> Any: 

153 return await self.list_renderer.render(request, resource) 

154 

155 

156class DetailActionHandler: 

157 def __init__(self, detail_renderer: Any): 

158 self.detail_renderer = detail_renderer 

159 

160 def can_handle(self, action: str) -> bool: 

161 return action == "detail" 

162 

163 async def handle(self, request: StarletteRequest, resource: Any, **kwargs) -> Any: 

164 item_id = request.path_params.get("id", "?") 

165 return await self.detail_renderer.render_detail(request, resource, item_id) 

166 

167 

168class CreateActionHandler: 

169 def __init__(self, form_renderer: Any): 

170 self.form_renderer = form_renderer 

171 

172 def can_handle(self, action: str) -> bool: 

173 return action == "create" 

174 

175 async def handle(self, request: StarletteRequest, resource: Any, **kwargs) -> Any: 

176 if request.method == "POST": 

177 return await self._handle_create(request, resource) 

178 return await self.form_renderer.render_create(request, resource) 

179 

180 async def _handle_create(self, request: StarletteRequest, resource: Any) -> Any: 

181 from lexigram.admin.resources.base import Resource 

182 

183 form = request.scope.get("admin_form_data") or await request.form() 

184 data = dict(form) 

185 data.pop("csrf_token", None) 

186 

187 if isinstance(resource, Resource) and resource._data_source: 

188 validation = await resource.before_validate(data) 

189 if validation.is_err(): 

190 error = validation.unwrap_err() 

191 return await self.form_renderer.render_create( 

192 request, resource, errors=_validation_errors_to_dict(error) 

193 ) 

194 

195 validated_data = validation.unwrap() 

196 validated = await resource.before_create(validated_data) 

197 record = await resource._data_source.create(validated) 

198 await resource.after_create(record) 

199 

200 resource_prefix = request.scope.get( 

201 "admin_resource_prefix", resource.name or "" 

202 ) 

203 

204 from starlette.responses import HTMLResponse 

205 

206 return HTMLResponse( 

207 f'<html><head><meta http-equiv="refresh" content="0;url=/admin/{resource_prefix}"></head><body></body></html>' 

208 ) 

209 

210 return await self.form_renderer.render_create(request, resource) 

211 

212 

213class EditActionHandler: 

214 def __init__(self, form_renderer: Any): 

215 self.form_renderer = form_renderer 

216 

217 def can_handle(self, action: str) -> bool: 

218 return action == "edit" 

219 

220 async def handle(self, request: StarletteRequest, resource: Any, **kwargs) -> Any: 

221 item_id = request.path_params.get("id", "?") 

222 if request.method == "POST": 

223 return await self._handle_update(request, resource, item_id) 

224 return await self.form_renderer.render_edit(request, resource, item_id) 

225 

226 async def _handle_update( 

227 self, request: StarletteRequest, resource: Any, item_id: str 

228 ) -> Any: 

229 from lexigram.admin.resources.base import Resource 

230 

231 form = request.scope.get("admin_form_data") or await request.form() 

232 data = dict(form) 

233 data.pop("csrf_token", None) 

234 

235 if isinstance(resource, Resource) and resource._data_source: 

236 validation = await resource.before_validate(data) 

237 if validation.is_err(): 

238 error = validation.unwrap_err() 

239 return await self.form_renderer.render_edit( 

240 request, 

241 resource, 

242 item_id, 

243 errors=_validation_errors_to_dict(error), 

244 ) 

245 

246 validated_data = validation.unwrap() 

247 validated = await resource.before_update(item_id, validated_data) 

248 record = await resource._data_source.find_one(item_id) 

249 can_update = getattr(resource, "can_update", None) 

250 if can_update and not can_update(record): 

251 return HTMLResponse("This record cannot be updated", status_code=403) 

252 updated_record = await resource._data_source.update(item_id, validated) 

253 await resource.after_update(updated_record) 

254 

255 resource_prefix = request.scope.get( 

256 "admin_resource_prefix", resource.name or "" 

257 ) 

258 

259 return HTMLResponse( 

260 f'<html><head><meta http-equiv="refresh" content="0;url=/admin/{resource_prefix}"></head><body></body></html>' 

261 ) 

262 

263 return await self.form_renderer.render_edit(request, resource, item_id) 

264 

265 

266class CloneActionHandler: 

267 """Handler for the ``clone`` action — duplicates a record and redirects.""" 

268 

269 def can_handle(self, action: str) -> bool: 

270 return action == "clone" 

271 

272 async def handle(self, request: StarletteRequest, resource: Any, **kwargs) -> Any: 

273 from lexigram.admin.resources.base import Resource 

274 

275 item_id = request.path_params.get("id", "?") 

276 

277 if not isinstance(resource, Resource): 

278 return HTMLResponse( 

279 "<h1>Clone not supported for this resource</h1>", status_code=400 

280 ) 

281 

282 new_record = await resource.duplicate(item_id) 

283 new_id = str(getattr(new_record, "id", "?")) 

284 resource_prefix = request.scope.get( 

285 "admin_resource_prefix", resource.name or "" 

286 ) 

287 from starlette.responses import RedirectResponse 

288 

289 return RedirectResponse( 

290 url=f"/admin/{resource_prefix}/{new_id}/edit", 

291 status_code=302, 

292 ) 

293 

294 

295class RestoreActionHandler: 

296 """Handler for the ``restore`` action — restores a soft-deleted record and redirects.""" 

297 

298 def can_handle(self, action: str) -> bool: 

299 return action == "restore" 

300 

301 async def handle(self, request: StarletteRequest, resource: Any, **kwargs) -> Any: 

302 from lexigram.admin.resources.base import Resource 

303 

304 item_id = request.path_params.get("id", "?") 

305 

306 if not isinstance(resource, Resource): 

307 return HTMLResponse( 

308 "<h1>Restore not supported for this resource</h1>", status_code=400 

309 ) 

310 

311 restored = await resource.restore(item_id) 

312 new_id = str(getattr(restored, "id", "?")) 

313 resource_prefix = request.scope.get( 

314 "admin_resource_prefix", resource.name or "" 

315 ) 

316 from starlette.responses import RedirectResponse 

317 

318 return RedirectResponse( 

319 url=f"/admin/{resource_prefix}/{new_id}/edit", 

320 status_code=302, 

321 ) 

322 

323 

324class PurgeActionHandler: 

325 """Handler for the ``purge`` action — permanently deletes a record and redirects.""" 

326 

327 def can_handle(self, action: str) -> bool: 

328 return action == "purge" 

329 

330 async def handle(self, request: StarletteRequest, resource: Any, **kwargs) -> Any: 

331 from lexigram.admin.resources.base import Resource 

332 

333 item_id = request.path_params.get("id", "?") 

334 

335 if not isinstance(resource, Resource): 

336 return HTMLResponse( 

337 "<h1>Purge not supported for this resource</h1>", status_code=400 

338 ) 

339 

340 await resource.purge(item_id) 

341 resource_prefix = request.scope.get( 

342 "admin_resource_prefix", resource.name or "" 

343 ) 

344 from starlette.responses import RedirectResponse 

345 

346 return RedirectResponse( 

347 url=f"/admin/{resource_prefix}", 

348 status_code=302, 

349 ) 

350 

351 

352class ImportActionHandler: 

353 """Handler for import download routes (example CSV, failed-import report). 

354 

355 Serves GET ``import-example`` (the resource's declared 

356 :class:`~lexigram.admin.actions.standard.ImportAction` template) and 

357 GET ``import-report`` (a stored failed-import report as CSV). 

358 """ 

359 

360 _ACTIONS = ("import-example", "import-report") 

361 

362 def can_handle(self, action: str) -> bool: 

363 """Whether this handler serves the given route action.""" 

364 return action in self._ACTIONS 

365 

366 @staticmethod 

367 def _find_import_action(resource: Any) -> Any: 

368 """Locate the resource's declared ImportAction, if any.""" 

369 from lexigram.admin.actions.standard import ImportAction 

370 

371 for collection in ("header_actions", "actions"): 

372 for action in getattr(resource, collection, None) or []: 

373 if isinstance(action, ImportAction): 

374 return action 

375 return None 

376 

377 @staticmethod 

378 def _csv_response(content: str, filename: str) -> Any: 

379 """Build an attachment CSV response.""" 

380 from starlette.responses import Response 

381 

382 return Response( 

383 content=content, 

384 media_type="text/csv", 

385 headers={"Content-Disposition": f'attachment; filename="{filename}"'}, 

386 ) 

387 

388 async def handle(self, request: StarletteRequest, resource: Any, **kwargs) -> Any: 

389 from starlette.responses import HTMLResponse 

390 

391 action = self._find_import_action(resource) 

392 if action is None: 

393 return HTMLResponse( 

394 "<h1>Import not configured for this resource</h1>", 

395 status_code=404, 

396 ) 

397 if request.method != "GET": 

398 return HTMLResponse("Method not allowed", status_code=405) 

399 

400 requested = request.scope.get("admin_action", "") 

401 if requested == "import-example": 

402 content = action.example_csv() 

403 if not content: 

404 return HTMLResponse( 

405 "<h1>No example CSV configured</h1>", status_code=404 

406 ) 

407 return self._csv_response(content, action.example_filename) 

408 

409 report_id = request.query_params.get("report_id", "") 

410 content = action.report_csv(report_id) 

411 if content is None: 

412 return HTMLResponse("<h1>Report not found</h1>", status_code=404) 

413 filename = action.report_filename(report_id) or "import-errors.csv" 

414 return self._csv_response(content, filename) 

415 

416 

417class DeleteActionHandler: 

418 """Handler for delete-confirm and delete actions.""" 

419 

420 def can_handle(self, action: str) -> bool: 

421 return action in ("delete-confirm", "delete") 

422 

423 async def handle(self, request: StarletteRequest, resource: Any, **kwargs) -> Any: 

424 item_id = request.path_params.get("id", "?") 

425 

426 if request.method == "GET": 

427 return await self._confirm_delete(request, resource, item_id) 

428 return await self._execute_delete(request, resource, item_id) 

429 

430 async def _confirm_delete( 

431 self, request: StarletteRequest, resource: Any, item_id: str 

432 ) -> Any: 

433 from lexigram.admin.resources.base import Resource as AdminResource 

434 

435 label = getattr(resource, "label", "Record") 

436 record_label = f"{label} #{item_id}" 

437 

438 if isinstance(resource, AdminResource) and resource._data_source: 

439 try: 

440 item = await resource._data_source.find_one(item_id) 

441 if item: 

442 for field in ("name", "title", "email", "username", "label"): 

443 val = ( 

444 item.get(field) 

445 if isinstance(item, dict) 

446 else getattr(item, field, None) 

447 ) 

448 if val: 

449 record_label = str(val) 

450 break 

451 except Exception: # noqa: S110 — intentional best-effort fallback 

452 pass 

453 

454 resource_prefix = request.scope.get("admin_resource_prefix", "") 

455 delete_url = f"/admin/{resource_prefix}/{item_id}/delete" 

456 

457 from lexigram.admin.ui.organisms.admin_slide_over import render_delete_confirm 

458 

459 html = render_delete_confirm( 

460 record_label=record_label, 

461 delete_url=delete_url, 

462 ) 

463 return HTMLResponse(html) 

464 

465 async def _execute_delete( 

466 self, request: StarletteRequest, resource: Any, item_id: str 

467 ) -> Any: 

468 from lexigram.admin.resources.base import Resource as AdminResource 

469 

470 if isinstance(resource, AdminResource) and resource._data_source: 

471 item = await resource._data_source.find_one(item_id) 

472 if item is None: 

473 return HTMLResponse("Not found", status_code=404) 

474 

475 can_delete = getattr(resource, "can_delete", None) 

476 if can_delete and not can_delete(item): 

477 is_htmx = request.headers.get("HX-Request") == "true" 

478 if is_htmx: 

479 response = HTMLResponse("") 

480 response.headers["HX-Trigger"] = ( 

481 '{"show-toast":{"message":"This record cannot be deleted","type":"error"}}' 

482 ) 

483 return response 

484 return HTMLResponse( 

485 '<html><head><meta http-equiv="refresh" content="0;url=/admin/"></head>' 

486 "<body>This record cannot be deleted</body></html>", 

487 status_code=409, 

488 ) 

489 

490 success = await resource._data_source.delete(item_id) 

491 if not success: 

492 return HTMLResponse("Not found", status_code=404) 

493 

494 after_delete = getattr(resource, "after_delete", None) 

495 if after_delete: 

496 await after_delete(item_id) 

497 

498 is_htmx = request.headers.get("HX-Request") == "true" 

499 resource_prefix = request.scope.get("admin_resource_prefix", "") 

500 

501 if is_htmx: 

502 response = HTMLResponse("") 

503 response.headers["HX-Trigger"] = ( 

504 '{"refresh-list":true,"show-toast":{"message":"Deleted successfully","type":"success"}}' 

505 ) 

506 response.headers["HX-Redirect"] = f"/admin/{resource_prefix}" 

507 return response 

508 

509 return HTMLResponse( 

510 f'<html><head><meta http-equiv="refresh" content="0;url=/admin/{resource_prefix}"></head><body></body></html>' 

511 ) 

512 

513 return HTMLResponse("Delete not supported", status_code=400) 

514 

515 

516class UserPermissionsActionHandler: 

517 """Handler for the per-user direct permission editing page (users only).""" 

518 

519 def __init__(self, config: AdminConfig) -> None: 

520 """Initialize the handler. 

521 

522 Args: 

523 config: Admin configuration (used for the CSRF secret). 

524 """ 

525 self._config = config 

526 

527 def can_handle(self, action: str) -> bool: 

528 """Whether this handler serves the given route action.""" 

529 return action == "permissions" 

530 

531 async def handle( 

532 self, request: StarletteRequest, resource: Any, **kwargs: Any 

533 ) -> Any: 

534 """Serve GET (form) and POST (save) for user permissions. 

535 

536 Only :class:`~lexigram.admin.resources.users.UserResource` 

537 instances support the page; other resources get a 404. 

538 """ 

539 from lexigram.admin.resources.users import UserResource 

540 

541 if not isinstance(resource, UserResource): 

542 return HTMLResponse( 

543 "<h1>Permissions not supported for this resource</h1>", 

544 status_code=404, 

545 ) 

546 item_id = request.path_params.get("id", "?") 

547 data_source = getattr(resource, "_data_source", None) 

548 if data_source is None: 

549 return HTMLResponse("Permissions not available", status_code=400) 

550 

551 if request.method == "POST": 

552 return await self._handle_submit(request, resource, data_source, item_id) 

553 return await self._handle_form(request, resource, data_source, item_id) 

554 

555 async def _handle_form( 

556 self, 

557 request: StarletteRequest, 

558 resource: Any, 

559 data_source: Any, 

560 item_id: str, 

561 ) -> Any: 

562 """Render the permission checkboxes for one user.""" 

563 user = await data_source.find_one(item_id) 

564 if user is None: 

565 return HTMLResponse("<h1>User not found</h1>", status_code=404) 

566 

567 self._ensure_csrf_token(request) 

568 prefix = request.scope.get("admin_resource_prefix", resource.name or "") 

569 

570 from lexigram.admin.resources.permissions_renderer import ( 

571 UserPermissionsRenderer, 

572 ) 

573 

574 renderer = UserPermissionsRenderer(resource_name=resource.name or "") 

575 return renderer.render_form( 

576 request=request, 

577 user=user, 

578 inventory=self._permission_inventory(resource), 

579 item_id=item_id, 

580 prefix=prefix, 

581 ) 

582 

583 async def _handle_submit( 

584 self, 

585 request: StarletteRequest, 

586 resource: Any, 

587 data_source: Any, 

588 item_id: str, 

589 ) -> Any: 

590 """Persist the submitted direct permissions for one user.""" 

591 form = request.scope.get("admin_form_data") or await request.form() 

592 permissions = sorted( 

593 {str(v).strip() for v in form.getlist("permissions") if str(v).strip()} 

594 ) 

595 

596 user = await data_source.find_one(item_id) 

597 prefix = request.scope.get("admin_resource_prefix", resource.name or "") 

598 if user is None: 

599 return RedirectResponse( 

600 url=f"/admin/{prefix}?error=User not found.", status_code=302 

601 ) 

602 await data_source.update(item_id, {"permissions": permissions}) 

603 return RedirectResponse( 

604 url=f"/admin/{prefix}?notice=User permissions updated.", 

605 status_code=302, 

606 ) 

607 

608 def _permission_inventory(self, resource: Any) -> Any: 

609 """Return the grouped permission inventory for the form. 

610 

611 Uses the inventory wired onto the resource at mount time; falls 

612 back to a local inventory scoped to the resource's own name. 

613 """ 

614 inventory = getattr(resource, "permission_inventory", None) 

615 if inventory is not None: 

616 return inventory 

617 from lexigram.admin.rbac.inventory import PermissionInventoryService 

618 

619 logger.debug( 

620 "admin.user_permissions_inventory_fallback", 

621 resource=resource.name, 

622 ) 

623 fallback = PermissionInventoryService() 

624 fallback.register_resources([resource.name or "users"]) 

625 return fallback 

626 

627 def _ensure_csrf_token(self, request: StarletteRequest) -> None: 

628 """Ensure ``request.state.csrf_token`` exists for the form embed.""" 

629 if getattr(getattr(request, "state", None), "csrf_token", None): 

630 return 

631 from lexigram.admin.auth.services.csrf_service import AdminCsrfService 

632 

633 session = getattr(request, "session", {}) 

634 session_id = session.get("csrf_session_id") or session.get( 

635 "admin_user_id", "anonymous" 

636 ) 

637 request.state.csrf_token = AdminCsrfService( 

638 secret=self._config.auth.session_secret.get_secret_value() 

639 ).generate_token(session_id) 

640 

641 

642class BulkActionHandler: 

643 """Handler for the ``bulk`` action — processes bulk operations.""" 

644 

645 _CONFIRM_LABELS = { 

646 "bulk-delete-confirm": ("delete", "Delete", "DELETE"), 

647 "bulk-purge-confirm": ("purge", "Purge", "PURGE"), 

648 "bulk-restore-confirm": ("restore", "Restore", "RESTORE"), 

649 } 

650 

651 def can_handle(self, action: str) -> bool: 

652 return action in ( 

653 "bulk", 

654 "bulk-delete-confirm", 

655 "bulk-purge-confirm", 

656 "bulk-restore-confirm", 

657 ) 

658 

659 async def handle(self, request: StarletteRequest, resource: Any, **kwargs) -> Any: 

660 from lexigram.admin.resources.base import Resource as AdminResource 

661 from lexigram.admin.ui.organisms.admin_slide_over import ( 

662 render_bulk_delete_confirm, 

663 ) 

664 

665 if not isinstance(resource, AdminResource) or not resource._data_source: 

666 return HTMLResponse( 

667 "<h1>Bulk actions not supported for this resource</h1>", 

668 status_code=400, 

669 ) 

670 

671 # ── Bulk confirmation (GET) ── 

672 if request.method == "GET": 

673 ids = request.query_params.getlist("ids") 

674 record_count = len(ids) 

675 resource_prefix = request.scope.get( 

676 "admin_resource_prefix", resource.name or "" 

677 ) 

678 bulk_url = f"/admin/{resource_prefix}/bulk" 

679 confirm_action = request.scope.get("admin_action", "bulk-delete-confirm") 

680 action, confirm_label, confirm_phrase = self._CONFIRM_LABELS.get( 

681 confirm_action, ("delete", "Delete", "DELETE") 

682 ) 

683 html = render_bulk_delete_confirm( 

684 record_count=record_count, 

685 bulk_url=bulk_url, 

686 action=action, 

687 confirm_label=confirm_label, 

688 confirm_phrase=confirm_phrase, 

689 ) 

690 return HTMLResponse(html) 

691 

692 # ── Bulk action execution (POST) ── 

693 form = request.scope.get("admin_form_data") 

694 if form is None: 

695 form = await request.form() 

696 action_name = form.get("action", "") 

697 form_ids = form.getlist("ids") if hasattr(form, "getlist") else [] 

698 

699 if not form_ids: 

700 return HTMLResponse("No records selected", status_code=400) 

701 

702 is_htmx = request.headers.get("HX-Request") == "true" 

703 

704 if action_name == "delete": 

705 count = await resource._data_source.bulk_delete(form_ids) 

706 message = f"Deleted {count} item(s)" 

707 elif action_name == "purge": 

708 count = await resource._data_source.bulk_delete(form_ids) 

709 message = f"Purged {count} item(s)" 

710 elif action_name == "restore": 

711 count = 0 

712 for item_id in form_ids: 

713 updated = await resource._data_source.update( 

714 item_id, {"deleted_at": None} 

715 ) 

716 if updated is not None: 

717 count += 1 

718 message = f"Restored {count} item(s)" 

719 else: 

720 return HTMLResponse(f"Unknown action: {action_name}", status_code=400) 

721 

722 if is_htmx: 

723 response = HTMLResponse(f"<p>{message}</p>") 

724 response.headers["HX-Trigger"] = ( 

725 '{"refresh-list":true,"show-toast":{"message":"' 

726 + message.replace('"', '\\"') 

727 + '","type":"success"}}' 

728 ) 

729 return response 

730 resource_prefix = request.scope.get( 

731 "admin_resource_prefix", resource.name or "" 

732 ) 

733 return RedirectResponse(url=f"/admin/{resource_prefix}", status_code=302) 

734 

735 

736class DefaultActionHandler: 

737 def can_handle(self, action: str) -> bool: 

738 return True 

739 

740 async def handle(self, request: StarletteRequest, resource: Any, **kwargs) -> Any: 

741 return HTMLResponse("<h1>Unknown action</h1>") 

742 

743 

744class ActionHandlerRegistry: 

745 """Registry for action handlers.""" 

746 

747 def __init__(self, config: AdminConfig, name: str, resources: dict | None = None): 

748 self._config = config 

749 self.name = name 

750 self._resources = resources or {} 

751 self._handlers: list[ResourceActionHandler] = [] 

752 self._initialize_handlers() 

753 

754 def _initialize_handlers(self) -> None: 

755 from lexigram.admin.engine.renderer import AdminRenderer as EngineAdminRenderer 

756 from lexigram.admin.resources.detail_renderer import DetailRenderer 

757 from lexigram.admin.resources.form_renderer import FormRenderer 

758 from lexigram.admin.resources.list_renderer import ListRenderer 

759 

760 # AdminRenderer is stateless — nav is resolved from request.app.state at 

761 # render time, so a fresh instance per handler registry is safe. 

762 renderer = EngineAdminRenderer() 

763 

764 list_renderer = ListRenderer(self._config, self.name, renderer) 

765 detail_renderer = DetailRenderer(self._config, self.name, renderer) 

766 form_renderer = FormRenderer(self._config, self.name, renderer) 

767 

768 self._handlers = [ 

769 ListActionHandler(list_renderer), 

770 DetailActionHandler(detail_renderer), 

771 CreateActionHandler(form_renderer), 

772 EditActionHandler(form_renderer), 

773 CloneActionHandler(), 

774 RestoreActionHandler(), 

775 PurgeActionHandler(), 

776 DeleteActionHandler(),