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

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: 

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.update(item_id, validated) 

249 await resource.after_update(record) 

250 

251 resource_prefix = request.scope.get( 

252 "admin_resource_prefix", resource.name or "" 

253 ) 

254 

255 from starlette.responses import HTMLResponse 

256 

257 return HTMLResponse( 

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

259 ) 

260 

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

262 

263 

264class CloneActionHandler: 

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

266 

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

268 return action == "clone" 

269 

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

271 from lexigram.admin.resources.base import Resource 

272 

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

274 

275 if not isinstance(resource, Resource): 

276 return HTMLResponse( 

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

278 ) 

279 

280 new_record = await resource.duplicate(item_id) 

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

282 resource_prefix = request.scope.get( 

283 "admin_resource_prefix", resource.name or "" 

284 ) 

285 from starlette.responses import RedirectResponse 

286 

287 return RedirectResponse( 

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

289 status_code=302, 

290 ) 

291 

292 

293class RestoreActionHandler: 

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

295 

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

297 return action == "restore" 

298 

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

300 from lexigram.admin.resources.base import Resource 

301 

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

303 

304 if not isinstance(resource, Resource): 

305 return HTMLResponse( 

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

307 ) 

308 

309 restored = await resource.restore(item_id) 

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

311 resource_prefix = request.scope.get( 

312 "admin_resource_prefix", resource.name or "" 

313 ) 

314 from starlette.responses import RedirectResponse 

315 

316 return RedirectResponse( 

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

318 status_code=302, 

319 ) 

320 

321 

322class PurgeActionHandler: 

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

324 

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

326 return action == "purge" 

327 

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

329 from lexigram.admin.resources.base import Resource 

330 

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

332 

333 if not isinstance(resource, Resource): 

334 return HTMLResponse( 

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

336 ) 

337 

338 await resource.purge(item_id) 

339 resource_prefix = request.scope.get( 

340 "admin_resource_prefix", resource.name or "" 

341 ) 

342 from starlette.responses import RedirectResponse 

343 

344 return RedirectResponse( 

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

346 status_code=302, 

347 ) 

348 

349 

350class DeleteActionHandler: 

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

352 

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

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

355 

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

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

358 

359 if request.method == "GET": 

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

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

362 

363 async def _confirm_delete( 

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

365 ) -> Any: 

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

367 

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

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

370 

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

372 try: 

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

374 if item: 

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

376 val = ( 

377 item.get(field) 

378 if isinstance(item, dict) 

379 else getattr(item, field, None) 

380 ) 

381 if val: 

382 record_label = str(val) 

383 break 

384 except Exception: 

385 pass 

386 

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

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

389 

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

391 

392 html = render_delete_confirm( 

393 record_label=record_label, 

394 delete_url=delete_url, 

395 ) 

396 return HTMLResponse(html) 

397 

398 async def _execute_delete( 

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

400 ) -> Any: 

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

402 

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

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

405 if not success: 

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

407 

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

409 if after_delete: 

410 await after_delete(item_id) 

411 

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

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

414 

415 if is_htmx: 

416 response = HTMLResponse("") 

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

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

419 ) 

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

421 return response 

422 

423 return HTMLResponse( 

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

425 ) 

426 

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

428 

429 

430class BulkActionHandler: 

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

432 

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

434 return action in ("bulk", "bulk-delete-confirm") 

435 

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

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

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

439 render_bulk_delete_confirm, 

440 ) 

441 

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

443 return HTMLResponse( 

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

445 status_code=400, 

446 ) 

447 

448 # ── Bulk delete confirmation (GET) ── 

449 if request.method == "GET": 

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

451 record_count = len(ids) 

452 resource_prefix = request.scope.get( 

453 "admin_resource_prefix", resource.name or "" 

454 ) 

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

456 html = render_bulk_delete_confirm( 

457 record_count=record_count, 

458 bulk_url=bulk_url, 

459 ) 

460 return HTMLResponse(html) 

461 

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

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

464 if form is None: 

465 form = await request.form() 

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

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

468 

469 if not form_ids: 

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

471 

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

473 

474 if action_name == "delete": 

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

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

477 

478 if is_htmx: 

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

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

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

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

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

484 ) 

485 return response 

486 resource_prefix = request.scope.get( 

487 "admin_resource_prefix", resource.name or "" 

488 ) 

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

490 

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

492 

493 

494class DefaultActionHandler: 

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

496 return True 

497 

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

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

500 

501 

502class ActionHandlerRegistry: 

503 """Registry for action handlers.""" 

504 

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

506 self._config = config 

507 self.name = name 

508 self._resources = resources or {} 

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

510 self._initialize_handlers() 

511 

512 def _initialize_handlers(self) -> None: 

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

514 from lexigram.admin.resources.detail_renderer import DetailRenderer 

515 from lexigram.admin.resources.form_renderer import FormRenderer 

516 from lexigram.admin.resources.list_renderer import ListRenderer 

517 

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

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

520 renderer = EngineAdminRenderer() 

521 

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

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

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

525 

526 self._handlers = [ 

527 ListActionHandler(list_renderer), 

528 DetailActionHandler(detail_renderer), 

529 CreateActionHandler(form_renderer), 

530 EditActionHandler(form_renderer), 

531 CloneActionHandler(), 

532 RestoreActionHandler(), 

533 PurgeActionHandler(), 

534 DeleteActionHandler(), 

535 BulkActionHandler(), 

536 DefaultActionHandler(), 

537 ] 

538 

539 async def handle( 

540 self, request: StarletteRequest, resource: Any, action: str 

541 ) -> Any: 

542 for handler in self._handlers: 

543 if handler.can_handle(action): 

544 return await handler.handle(request, resource) 

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

546 

547 

548@inject 

549class ResourceHandler: 

550 """Handler for resource routes.""" 

551 

552 def __init__( 

553 self, 

554 config: AdminConfig, 

555 name: str, 

556 action: str, 

557 resources: dict | None = None, 

558 ): 

559 self._config = config 

560 self.name = name 

561 self.action = action 

562 self._resources = resources or {} 

563 self._registry = ActionHandlerRegistry(config, name, resources=self._resources) 

564 

565 async def __call__(self, scope, receive, send) -> Any: 

566 request = StarletteRequest(scope, receive, send) 

567 scope["admin_resource_prefix"] = self.name 

568 resource = self._resources.get(self.name) if self._resources else None 

569 response = await self._registry.handle(request, resource, self.action) 

570 await response(scope, receive, send)