Coverage for src / lexigram / admin / controllers / resource.py: 0%
304 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 18:58 +0800
1"""Resource controller for CRUD operations.
3Provides a base controller class for handling resource CRUD with
4HTMX support, automatic pagination, filtering, and bulk actions.
5"""
7from __future__ import annotations
9from abc import ABC
10from dataclasses import dataclass
11from datetime import UTC, datetime
12from typing import Any, Generic, TypeVar
14from starlette.requests import Request
15from starlette.responses import HTMLResponse, RedirectResponse, Response
17from lexigram.admin.data.data_source import IDataSource as DataSourceProtocol
18from lexigram.admin.data.data_source import QueryResult
19from lexigram.admin.data.query import QuerySpec
20from lexigram.admin.exceptions import AdminValidationError, NotFoundError
21from lexigram.admin.state.context import AdminContext, AdminContextManager
22from lexigram.admin.state.url import URLState
23from lexigram.admin.ui.organisms.admin_slide_over import (
24 render_bulk_delete_confirm,
25 render_delete_confirm,
26)
27from lexigram.di.decorators import inject
28from lexigram.logging import get_logger
29from lexigram.ui import el, render_to_string
31logger = get_logger(__name__)
33T = TypeVar("T")
36@dataclass
37class ResourceMeta:
38 """Metadata for a resource."""
40 name: str
41 label: str
42 label_plural: str
43 icon: str | None = None
45 # URLs
46 prefix: str = ""
48 # Default settings
49 per_page: int = 20
50 searchable_fields: list[str] | None = None
51 default_sort: str = "id"
52 default_sort_order: str = "desc"
54 # Features
55 enable_create: bool = True
56 enable_edit: bool = True
57 enable_delete: bool = True
58 enable_clone: bool = True
59 enable_bulk_actions: bool = True
60 enable_export: bool = True
62 @classmethod
63 def from_dict(cls, data: dict[str, Any]) -> ResourceMeta:
64 """Create from dictionary."""
65 return cls(
66 name=data.get("name", ""),
67 label=data.get("label", ""),
68 label_plural=data.get("label_plural", ""),
69 icon=data.get("icon"),
70 prefix=data.get("prefix", ""),
71 per_page=data.get("per_page", 20),
72 searchable_fields=data.get("searchable_fields"),
73 default_sort=data.get("default_sort", "id"),
74 default_sort_order=data.get("default_sort_order", "desc"),
75 enable_create=data.get("enable_create", True),
76 enable_edit=data.get("enable_edit", True),
77 enable_delete=data.get("enable_delete", True),
78 enable_clone=data.get("enable_clone", True),
79 enable_bulk_actions=data.get("enable_bulk_actions", True),
80 enable_export=data.get("enable_export", True),
81 )
84@inject
85class ResourceController(ABC, Generic[T]):
86 """Base controller for resource CRUD operations.
88 Provides standard CRUD endpoints with HTMX support:
89 - GET /{resource} - List with pagination/filtering
90 - GET /{resource}/{id} - View single resource
91 - GET /{resource}/create - Create form
92 - POST /{resource} - Create resource
93 - GET /{resource}/{id}/edit - Edit form
94 - PUT /{resource}/{id} - Update resource
95 - DELETE /{resource}/{id} - Delete resource
96 - POST /{resource}/bulk - Bulk actions
98 Subclasses should implement:
99 - get_data_source() - Return DataSourceProtocol for this resource
100 - get_columns() - Return columns for list view
101 - render_list() - Render list view HTML
102 - render_detail() - Render detail view HTML
103 - render_form() - Render create/edit form HTML
104 """
106 meta: ResourceMeta
108 # When True, DELETE calls soft-delete (sets deleted_at) instead of hard-delete.
109 # Use RepositoryDataSource(soft_delete_enabled=True) to filter them in queries.
110 soft_delete_enabled: bool = False
112 def __init__(
113 self,
114 data_source: DataSourceProtocol[T] | None = None,
115 meta: ResourceMeta | None = None,
116 ):
117 self._data_source = data_source
118 if meta:
119 self.meta = meta
120 # Optional audit logger — set via set_audit_logger() or DI
121 self._audit_logger: Any = None
122 # Optional revision service — set via set_revision_service() or DI
123 self._revision_service: Any = None
125 def set_audit_logger(self, audit_logger: Any) -> None:
126 """Attach an audit logger for CRUD event tracking.
128 Args:
129 audit_logger: Any object implementing ``async log(AuditEntry) -> None``.
130 """
131 self._audit_logger = audit_logger
133 def set_revision_service(self, revision_service: Any) -> None:
134 """Attach a :class:`~lexigram.admin.services.revisions.RevisionService`.
136 When set, a snapshot is recorded after every successful create or
137 update. The service also exposes ``diff`` and ``revert_data`` for the
138 revision history UI.
140 Args:
141 revision_service: ``RevisionService`` instance.
142 """
143 self._revision_service = revision_service
145 async def _record_revision(
146 self,
147 request: Request,
148 resource_id: str,
149 data: dict[str, Any],
150 comment: str = "",
151 ) -> None:
152 """Silently create a revision snapshot if a service is attached.
154 Args:
155 request: Current HTTP request (actor identity extracted from state).
156 resource_id: Record identifier.
157 data: Full field snapshot to persist.
158 comment: Optional human-readable description.
159 """
160 if self._revision_service is None:
161 return
162 try:
163 user = getattr(request.state, "user", None)
164 actor_id = str(getattr(user, "id", getattr(user, "user_id", "system")))
165 resource_name = getattr(self.meta, "name", "resource")
166 await self._revision_service.record(
167 resource_name,
168 resource_id,
169 data,
170 actor_id=actor_id,
171 comment=comment,
172 )
173 except Exception: # noqa: BLE001
174 logger.warning(
175 "Revision recording failed for resource %s/%s",
176 getattr(self.meta, "name", ""),
177 resource_id,
178 )
180 async def _emit_audit(
181 self,
182 request: Request,
183 action: str,
184 item_id: str = "",
185 old_values: dict[str, Any] | None = None,
186 new_values: dict[str, Any] | None = None,
187 outcome: str = "success",
188 ) -> None:
189 """Emit an audit log entry if a logger is attached.
191 Args:
192 request: Current HTTP request (used for actor identity and metadata).
193 action: Machine-readable action (e.g. ``"user.create"``).
194 item_id: Identifier of the affected record.
195 old_values: Field snapshot before the change.
196 new_values: Field snapshot after the change.
197 outcome: ``"success"`` or ``"failure"``.
198 """
199 if self._audit_logger is None:
200 return
201 try:
202 from lexigram.contracts.audit import AuditEntry
204 user = getattr(request.state, "user", None)
205 actor_id = str(getattr(user, "id", getattr(user, "user_id", "anonymous")))
206 resource_name = getattr(self.meta, "name", "unknown")
207 entry = AuditEntry(
208 action=action,
209 actor_id=actor_id,
210 resource_type=resource_name,
211 resource_id=str(item_id),
212 outcome=outcome,
213 old_values=old_values,
214 new_values=new_values,
215 metadata={
216 "ip": request.client.host if request.client else "",
217 "user_agent": request.headers.get("user-agent", ""),
218 "path": str(request.url),
219 },
220 )
221 await self._audit_logger.log(entry)
222 except Exception: # noqa: BLE001
223 logger.warning(
224 "Audit logging failed for action %s on %s/%s",
225 action,
226 getattr(self.meta, "name", ""),
227 item_id,
228 )
230 def get_data_source(self) -> DataSourceProtocol[T]:
231 """Get the data source for this resource."""
232 if self._data_source is None:
233 raise NotImplementedError("Subclass must implement get_data_source()")
234 return self._data_source
236 # ========== List ==========
238 async def list_view(self, request: Request) -> Response:
239 """List resources with pagination and filtering."""
240 async with AdminContextManager(request) as ctx:
241 # Parse URL state
242 url_state = URLState.from_request(request)
244 # Build query
245 query = self._build_query(url_state)
247 # Fetch data
248 data_source = self.get_data_source()
249 result = await data_source.find_many(query)
251 # Check if HTMX request
252 if ctx.is_htmx:
253 # Return just the table/content
254 return HTMLResponse(self.render_list_partial(ctx, result, url_state))
256 # Return full page
257 return HTMLResponse(self.render_list(ctx, result, url_state))
259 def _build_query(self, state: URLState) -> QuerySpec:
260 """Build QuerySpec from URL state."""
261 qs = QuerySpec()
263 # Pagination — cursor takes priority over page number
264 if state.cursor:
265 qs = qs.with_cursor(state.cursor).with_per_page(
266 state.per_page or self.meta.per_page
267 )
268 else:
269 qs = qs.with_page(state.page).with_per_page(
270 state.per_page or self.meta.per_page
271 )
273 # Sorting
274 if state.sort:
275 qs = qs.with_order_by(state.sort, state.order)
276 else:
277 qs = qs.with_order_by(self.meta.default_sort, self.meta.default_sort_order)
279 # Search
280 if state.search:
281 qs = qs.with_search(state.search, self.meta.searchable_fields or [])
283 # Filters
284 for field, value in state.filters.items():
285 if isinstance(value, list):
286 qs = qs.with_where_in(field, value)
287 else:
288 qs = qs.with_where_eq(field, value)
290 return qs
292 # ========== Detail ==========
294 async def detail(self, request: Request) -> Response:
295 """View single resource."""
296 async with AdminContextManager(request) as ctx:
297 item_id = request.path_params.get("id")
299 data_source = self.get_data_source()
300 item = await data_source.find_one(item_id)
302 if item is None:
303 from lexigram.admin.lib.template import render_error_page
305 html = render_error_page(
306 status_code=404,
307 title="Not Found",
308 message=f"{self.meta.label} not found",
309 )
310 return HTMLResponse(html, status_code=404)
312 if ctx.is_htmx:
313 return HTMLResponse(self.render_detail_partial(ctx, item))
315 return HTMLResponse(self.render_detail(ctx, item))
317 # ========== Create ==========
319 async def create_form(self, request: Request) -> Response:
320 """Show create form."""
321 async with AdminContextManager(request) as ctx:
322 if ctx.is_htmx:
323 return HTMLResponse(self.render_form_partial(ctx, None))
324 return HTMLResponse(self.render_form(ctx, None))
326 async def create(self, request: Request) -> Response:
327 """Create new resource."""
328 async with AdminContextManager(request) as ctx:
329 form_data = request.scope.get("admin_form_data")
330 if form_data is None:
331 form_data = await request.form()
332 data = dict(form_data)
334 # Validate
335 try:
336 validated = self.validate_create(data)
337 except AdminValidationError as e:
338 # Return form with errors
339 if ctx.is_htmx:
340 return HTMLResponse(
341 self.render_form_partial(ctx, None, data, e.details or {}),
342 status_code=422,
343 )
344 return HTMLResponse(
345 self.render_form(ctx, None, data, e.details or {}),
346 status_code=422,
347 )
349 # Create
350 data_source = self.get_data_source()
351 created = await data_source.create(validated)
352 created_id = str(getattr(created, "id", "") if created else "")
353 await self._emit_audit(
354 request,
355 f"{getattr(self.meta, 'name', 'resource')}.create",
356 item_id=created_id,
357 new_values=validated,
358 )
359 await self._record_revision(
360 request, created_id, validated, comment="create"
361 )
363 # Redirect or return success
364 ctx.add_flash("Created successfully", "success")
365 if ctx.is_htmx:
366 response = Response(status_code=200)
367 response.headers["HX-Redirect"] = f"{self.meta.prefix}/{self.meta.name}"
368 return response
370 return RedirectResponse(
371 url=f"{self.meta.prefix}/{self.meta.name}",
372 status_code=302,
373 )
375 def validate_create(self, data: dict[str, Any]) -> dict[str, Any]:
376 """Validate data for create. Override to customize."""
377 return data
379 # ========== Update ==========
381 async def edit_form(self, request: Request) -> Response:
382 """Show edit form."""
383 async with AdminContextManager(request) as ctx:
384 item_id = request.path_params.get("id")
386 data_source = self.get_data_source()
387 item = await data_source.find_one(item_id)
389 if item is None:
390 raise NotFoundError(message=f"{self.meta.label} not found")
392 if ctx.is_htmx:
393 return HTMLResponse(self.render_form_partial(ctx, item))
394 return HTMLResponse(self.render_form(ctx, item))
396 async def update(self, request: Request) -> Response:
397 """Update existing resource."""
398 async with AdminContextManager(request) as ctx:
399 item_id = request.path_params.get("id")
400 form_data = request.scope.get("admin_form_data")
401 if form_data is None:
402 form_data = await request.form()
403 data = dict(form_data)
405 # Validate
406 try:
407 validated = self.validate_update(item_id, data)
408 except AdminValidationError as e:
409 # Get current item for form
410 item = await self.get_data_source().find_one(item_id)
411 if ctx.is_htmx:
412 return HTMLResponse(
413 self.render_form_partial(ctx, item, data, e.details or {}),
414 status_code=422,
415 )
416 return HTMLResponse(
417 self.render_form(ctx, item, data, e.details or {}),
418 status_code=422,
419 )
421 # Update
422 data_source = self.get_data_source()
423 item = await data_source.update(item_id, validated)
424 await self._emit_audit(
425 request,
426 f"{getattr(self.meta, 'name', 'resource')}.update",
427 item_id=str(item_id),
428 new_values=validated,
429 )
430 await self._record_revision(
431 request, str(item_id), validated, comment="update"
432 )
434 ctx.add_flash("Updated successfully", "success")
435 if ctx.is_htmx:
436 response = Response(status_code=200)
437 response.headers["HX-Redirect"] = (
438 f"{self.meta.prefix}/{self.meta.name}/{item_id}"
439 )
440 return response
442 return RedirectResponse(
443 url=f"{self.meta.prefix}/{self.meta.name}/{item_id}",
444 status_code=302,
445 )
447 def validate_update(self, item_id: Any, data: dict[str, Any]) -> dict[str, Any]:
448 """Validate data for update. Override to customize."""
449 return data
451 # ========== Delete ==========
453 async def delete_confirm(self, request: Request) -> Response:
454 """Render a delete confirmation slide-over panel.
456 Called via HTMX GET from the Delete row action. Returns an
457 AdminSlideOver fragment that lets the user confirm or cancel
458 the deletion without using the native browser confirm dialog.
459 """
460 item_id = request.path_params.get("id")
461 label = self.meta.label
463 # Attempt to fetch a human-readable label for the record
464 record_label = f"{label} #{item_id}"
465 try:
466 data_source = self.get_data_source()
467 item = await data_source.find_one(item_id)
468 if item:
469 for field in ("name", "title", "email", "username", "label"):
470 val = getattr(item, field, None)
471 if val:
472 record_label = str(val)
473 break
474 except Exception: # noqa: BLE001
475 pass
477 delete_url = f"{self.meta.prefix}/{self.meta.name}/{item_id}"
478 html = render_delete_confirm(
479 record_label=record_label,
480 delete_url=delete_url,
481 )
482 return HTMLResponse(html)
484 async def bulk_delete_confirm(self, request: Request) -> Response:
485 """Render a bulk delete confirmation slide-over panel.
487 Called via HTMX GET from a BulkAction button. Reads the selected
488 record IDs from the query string (passed via ``hx-include`` of the
489 checked checkboxes) and renders a slide-over confirmation panel.
490 """
491 ids = request.query_params.getlist("ids")
492 record_count = len(ids)
494 bulk_url = f"{self.meta.prefix}/{self.meta.name}/bulk"
495 html = render_bulk_delete_confirm(
496 record_count=record_count,
497 bulk_url=bulk_url,
498 )
499 return HTMLResponse(html)
501 async def delete(self, request: Request) -> Response:
502 """Delete resource (soft or hard depending on soft_delete_enabled)."""
503 async with AdminContextManager(request) as ctx:
504 item_id = request.path_params.get("id")
505 data_source = self.get_data_source()
507 if self.soft_delete_enabled:
508 # Soft delete — stamp deleted_at instead of removing the row
509 updated = await data_source.update(
510 item_id, {"deleted_at": datetime.now(UTC).isoformat()}
511 )
512 if updated is None:
513 raise NotFoundError(message=f"{self.meta.label} not found")
514 await self._emit_audit(
515 request,
516 f"{getattr(self.meta, 'name', 'resource')}.soft_delete",
517 item_id=str(item_id),
518 )
519 else:
520 success = await data_source.delete(item_id)
521 if not success:
522 raise NotFoundError(message=f"{self.meta.label} not found")
523 await self._emit_audit(
524 request,
525 f"{getattr(self.meta, 'name', 'resource')}.delete",
526 item_id=str(item_id),
527 )
529 ctx.add_flash("Deleted successfully", "success")
530 if ctx.is_htmx:
531 response = Response(status_code=200)
532 response.headers["HX-Redirect"] = f"{self.meta.prefix}/{self.meta.name}"
533 return response
535 return RedirectResponse(
536 url=f"{self.meta.prefix}/{self.meta.name}",
537 status_code=302,
538 )
540 async def restore(self, request: Request) -> Response:
541 """Restore a soft-deleted resource (clears deleted_at).
543 Only available when soft_delete_enabled is True.
544 """
545 async with AdminContextManager(request) as ctx:
546 if not self.soft_delete_enabled:
547 return HTMLResponse(
548 "Soft delete is not enabled for this resource", status_code=400
549 )
551 item_id = request.path_params.get("id")
552 data_source = self.get_data_source()
554 updated = await data_source.update(item_id, {"deleted_at": None})
555 if updated is None:
556 raise NotFoundError(message=f"{self.meta.label} not found")
557 await self._emit_audit(
558 request,
559 f"{getattr(self.meta, 'name', 'resource')}.restore",
560 item_id=str(item_id),
561 )
563 ctx.add_flash("Restored successfully", "success")
564 if ctx.is_htmx:
565 response = Response(status_code=200)
566 response.headers["HX-Trigger"] = (
567 '{"refresh-list":true,"show-toast":{"message":"Restored successfully","type":"success"}}'
568 )
569 return response
571 return RedirectResponse(
572 url=f"{self.meta.prefix}/{self.meta.name}",
573 status_code=302,
574 )
576 # ========== Bulk Actions ==========
578 async def bulk_action(self, request: Request) -> Response:
579 """Handle bulk actions."""
580 async with AdminContextManager(request) as ctx:
581 form_data = request.scope.get("admin_form_data")
582 if form_data is None:
583 form_data = await request.form()
584 action = form_data.get("action")
585 ids = form_data.getlist("ids")
587 if not action or not ids:
588 return HTMLResponse("Missing action or ids", status_code=400)
590 result = await self.execute_bulk_action(action, ids) # type: ignore[arg-type]
592 ctx.add_flash(result, "success")
593 if ctx.is_htmx:
594 response = HTMLResponse(render_to_string(el("p", str(result))))
595 response.headers["HX-Trigger"] = (
596 '{"refresh-list":true,"show-toast":{"message":"'
597 + result.replace('"', '\\"')
598 + '","type":"success"}}'
599 )
600 return response
602 return RedirectResponse(
603 url=f"{self.meta.prefix}/{self.meta.name}",
604 status_code=302,
605 )
607 async def execute_bulk_action(self, action: str, ids: list[str]) -> str:
608 """Execute bulk action. Override to add custom actions.
610 When the record count meets or exceeds the configured
611 ``bulk_threshold`` (from ``TasksIntegrationConfig``), the action is
612 dispatched through the tasks integration instead of running inline.
613 """
614 if self._should_dispatch_via_tasks(len(ids)):
615 return await self._dispatch_via_tasks(action, ids)
617 data_source = self.get_data_source()
619 if action == "delete":
620 count = await data_source.bulk_delete(ids)
621 return f"Deleted {count} items"
623 return f"Unknown action: {action}"
625 def _should_dispatch_via_tasks(self, count: int) -> bool:
626 """Check if the bulk count exceeds the tasks threshold."""
627 from lexigram.admin.integrations import get as get_integration
629 tasks = get_integration("TasksIntegration")
630 if not tasks:
631 return False
632 if not tasks._enabled:
633 return False
634 return count >= tasks.threshold
636 async def _dispatch_via_tasks(self, action: str, ids: list[str]) -> str:
637 """Dispatch a bulk action through the tasks integration."""
638 from lexigram.admin.integrations import get as get_integration
640 tasks = get_integration("TasksIntegration")
641 if not tasks:
642 return "Task system unavailable"
644 result = await tasks.dispatch(
645 runner=action,
646 action_name=action,
647 record_ids=ids,
648 ctx_summary=f"Bulk {action} of {len(ids)} records",
649 )
650 return f"Scheduled bulk {action} for {len(ids)} records (task: {result.get('status', 'unknown')})"
652 # ========== Rendering (Override these) ==========
654 def render_list(
655 self,
656 ctx: AdminContext,
657 result: QueryResult[T],
658 state: URLState,
659 ) -> str:
660 """Render full list page. Override in subclass."""
661 return f"""
662<!DOCTYPE html>
663<html>
664<head>
665 <title>{self.meta.label_plural}</title>
666 <script src="https://unpkg.com/htmx.org@1.9.10"></script>
667</head>
668<body>
669 <h1>{self.meta.label_plural}</h1>
670 {self.render_list_partial(ctx, result, state)}
671</body>
672</html>
673"""
675 def render_list_partial(
676 self,
677 ctx: AdminContext,
678 result: QueryResult[T],
679 state: URLState,
680 ) -> str:
681 """Render list content (for HTMX). Override in subclass."""
682 row_els: list[Any] = []
683 for item in result.items:
684 row_els.append(el("tr", el("td", str(item))))
686 total_pages = getattr(result, "total_pages", "?")
687 return f"""
688<table>
689 <thead><tr><th>Item</th></tr></thead>
690 {render_to_string(el("tbody", *row_els))}
691</table>
692<div>Page {result.page} of {total_pages}</div>
693"""
695 def render_detail(self, ctx: AdminContext, item: T) -> str:
696 """Render full detail page. Override in subclass."""
697 return f"""
698<!DOCTYPE html>
699<html>
700<head><title>{self.meta.label}</title></head>
701<body>
702 <h1>{self.meta.label}</h1>
703 {self.render_detail_partial(ctx, item)}
704</body>
705</html>
706"""
708 def render_detail_partial(self, ctx: AdminContext, item: T) -> str:
709 """Render detail content. Override in subclass."""
710 return render_to_string(el("pre", str(item)))
712 def render_form(
713 self,
714 ctx: AdminContext,
715 item: T | None,
716 data: dict[str, Any] | None = None,
717 errors: dict[str, list[str]] | None = None,
718 ) -> str:
719 """Render full form page. Override in subclass."""
720 title = f"Edit {self.meta.label}" if item else f"Create {self.meta.label}"
721 return f"""
722<!DOCTYPE html>
723<html>
724<head><title>{title}</title></head>
725<body>
726 <h1>{title}</h1>
727 {self.render_form_partial(ctx, item, data, errors)}
728</body>
729</html>
730"""
732 def render_form_partial(
733 self,
734 ctx: AdminContext,
735 item: T | None,
736 data: dict[str, Any] | None = None,
737 errors: dict[str, list[str]] | None = None,
738 ) -> str:
739 """Render form content. Override in subclass."""
740 action = f"{self.meta.prefix}/{self.meta.name}"
741 if item:
742 id_val = getattr(item, "id", None)
743 action = f"{action}/{id_val}"
745 return f"""
746<form method="POST" action="{action}">
747 <p>Override render_form_partial() to customize</p>
748 <button type="submit">Save</button>
749</form>
750"""
752 # ========== Route Registration ==========
754 def get_routes(self) -> list:
755 """Get Starlette routes for this controller."""
756 from starlette.routing import Route
758 prefix = f"/{self.meta.name}"
760 return [
761 Route(prefix, self.list_view, methods=["GET"]),
762 Route(f"{prefix}/create", self.create_form, methods=["GET"]),
763 Route(prefix, self.create, methods=["POST"]),
764 Route(f"{prefix}/bulk", self.bulk_action, methods=["POST"]),
765 Route(
766 f"{prefix}/bulk-delete-confirm",
767 self.bulk_delete_confirm,
768 methods=["GET"],
769 ),
770 Route(f"{prefix}/{{id}}", self.detail, methods=["GET"]),
771 Route(f"{prefix}/{{id}}/edit", self.edit_form, methods=["GET"]),
772 Route(
773 f"{prefix}/{{id}}/delete-confirm", self.delete_confirm, methods=["GET"]
774 ),
775 Route(f"{prefix}/{{id}}", self.update, methods=["PUT", "POST"]),
776 Route(f"{prefix}/{{id}}", self.delete, methods=["DELETE"]),
777 ]