Coverage for src/lexigram/admin/resources/handler.py: 0%
147 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1from __future__ import annotations
3from typing import Any
5from starlette.requests import Request as StarletteRequest
6from starlette.responses import HTMLResponse, RedirectResponse
8from lexigram.admin.config import AdminConfig
9from lexigram.di.decorators import inject
10from lexigram.logging import get_logger
12logger = get_logger(__name__)
14from lexigram.admin.resources.action_handlers import (
15 CloneActionHandler,
16 CreateActionHandler,
17 DeleteActionHandler,
18 DetailActionHandler,
19 EditActionHandler,
20 ImportActionHandler,
21 ListActionHandler,
22 PurgeActionHandler,
23 ResourceActionHandler,
24 RestoreActionHandler,
25)
28class UserPermissionsActionHandler:
29 """Handler for the per-user direct permission editing page (users only)."""
31 def __init__(self, config: AdminConfig) -> None:
32 """Initialize the handler.
34 Args:
35 config: Admin configuration (used for the CSRF secret).
36 """
37 self._config = config
39 def can_handle(self, action: str) -> bool:
40 """Whether this handler serves the given route action."""
41 return action == "permissions"
43 async def handle(
44 self, request: StarletteRequest, resource: Any, **kwargs: Any
45 ) -> Any:
46 """Serve GET (form) and POST (save) for user permissions.
48 Only :class:`~lexigram.admin.resources.users.UserResource`
49 instances support the page; other resources get a 404.
50 """
51 from lexigram.admin.resources.users import UserResource
53 if not isinstance(resource, UserResource):
54 return HTMLResponse(
55 "<h1>Permissions not supported for this resource</h1>",
56 status_code=404,
57 )
58 item_id = request.path_params.get("id", "?")
59 data_source = getattr(resource, "_data_source", None)
60 if data_source is None:
61 return HTMLResponse("Permissions not available", status_code=400)
63 if request.method == "POST":
64 return await self._handle_submit(request, resource, data_source, item_id)
65 return await self._handle_form(request, resource, data_source, item_id)
67 async def _handle_form(
68 self,
69 request: StarletteRequest,
70 resource: Any,
71 data_source: Any,
72 item_id: str,
73 ) -> Any:
74 """Render the permission checkboxes for one user."""
75 user = await data_source.find_one(item_id)
76 if user is None:
77 return HTMLResponse("<h1>User not found</h1>", status_code=404)
79 self._ensure_csrf_token(request)
80 prefix = request.scope.get("admin_resource_prefix", resource.name or "")
82 from lexigram.admin.resources.permissions_renderer import (
83 UserPermissionsRenderer,
84 )
86 renderer = UserPermissionsRenderer(resource_name=resource.name or "")
87 return renderer.render_form(
88 request=request,
89 user=user,
90 inventory=self._permission_inventory(resource),
91 item_id=item_id,
92 prefix=prefix,
93 )
95 async def _handle_submit(
96 self,
97 request: StarletteRequest,
98 resource: Any,
99 data_source: Any,
100 item_id: str,
101 ) -> Any:
102 """Persist the submitted direct permissions for one user."""
103 form = request.scope.get("admin_form_data") or await request.form()
104 permissions = sorted(
105 {str(v).strip() for v in form.getlist("permissions") if str(v).strip()}
106 )
108 user = await data_source.find_one(item_id)
109 prefix = request.scope.get("admin_resource_prefix", resource.name or "")
110 if user is None:
111 return RedirectResponse(
112 url=f"/admin/{prefix}?error=User not found.", status_code=302
113 )
114 await data_source.update(item_id, {"permissions": permissions})
115 return RedirectResponse(
116 url=f"/admin/{prefix}?notice=User permissions updated.",
117 status_code=302,
118 )
120 def _permission_inventory(self, resource: Any) -> Any:
121 """Return the grouped permission inventory for the form.
123 Uses the inventory wired onto the resource at mount time; falls
124 back to a local inventory scoped to the resource's own name.
125 """
126 inventory = getattr(resource, "permission_inventory", None)
127 if inventory is not None:
128 return inventory
129 from lexigram.admin.rbac.inventory import PermissionInventoryService
131 logger.debug(
132 "admin.user_permissions_inventory_fallback",
133 resource=resource.name,
134 )
135 fallback = PermissionInventoryService()
136 fallback.register_resources([resource.name or "users"])
137 return fallback
139 def _ensure_csrf_token(self, request: StarletteRequest) -> None:
140 """Ensure ``request.state.csrf_token`` exists for the form embed."""
141 if getattr(getattr(request, "state", None), "csrf_token", None):
142 return
143 from lexigram.admin.auth.services.csrf_service import AdminCsrfService
145 session = getattr(request, "session", {})
146 session_id = session.get("csrf_session_id") or session.get(
147 "admin_user_id", "anonymous"
148 )
149 request.state.csrf_token = AdminCsrfService(
150 secret=self._config.auth.session_secret.get_secret_value()
151 ).generate_token(session_id)
154class BulkActionHandler:
155 """Handler for the ``bulk`` action — processes bulk operations."""
157 _CONFIRM_LABELS = {
158 "bulk-delete-confirm": ("delete", "Delete", "DELETE"),
159 "bulk-purge-confirm": ("purge", "Purge", "PURGE"),
160 "bulk-restore-confirm": ("restore", "Restore", "RESTORE"),
161 }
163 def can_handle(self, action: str) -> bool:
164 return action in (
165 "bulk",
166 "bulk-delete-confirm",
167 "bulk-purge-confirm",
168 "bulk-restore-confirm",
169 )
171 async def handle(
172 self, request: StarletteRequest, resource: Any, **kwargs: Any
173 ) -> Any:
174 from lexigram.admin.resources.base import Resource as AdminResource
175 from lexigram.admin.ui.organisms.admin_slide_over import (
176 render_bulk_delete_confirm,
177 )
179 if not isinstance(resource, AdminResource) or not resource._data_source:
180 return HTMLResponse(
181 "<h1>Bulk actions not supported for this resource</h1>",
182 status_code=400,
183 )
185 # ── Bulk confirmation (GET) ──
186 if request.method == "GET":
187 ids = request.query_params.getlist("ids")
188 record_count = len(ids)
189 resource_prefix = request.scope.get(
190 "admin_resource_prefix", resource.name or ""
191 )
192 bulk_url = f"/admin/{resource_prefix}/bulk"
193 confirm_action = request.scope.get("admin_action", "bulk-delete-confirm")
194 action, confirm_label, confirm_phrase = self._CONFIRM_LABELS.get(
195 confirm_action, ("delete", "Delete", "DELETE")
196 )
197 html = render_bulk_delete_confirm(
198 record_count=record_count,
199 bulk_url=bulk_url,
200 action=action,
201 confirm_label=confirm_label,
202 confirm_phrase=confirm_phrase,
203 )
204 return HTMLResponse(html)
206 # ── Bulk action execution (POST) ──
207 form = request.scope.get("admin_form_data")
208 if form is None:
209 form = await request.form()
210 action_name = form.get("action", "")
211 form_ids = form.getlist("ids") if hasattr(form, "getlist") else []
213 if not form_ids:
214 return HTMLResponse("No records selected", status_code=400)
216 is_htmx = request.headers.get("HX-Request") == "true"
218 if action_name == "delete":
219 count = await resource._data_source.bulk_delete(form_ids)
220 message = f"Deleted {count} item(s)"
221 elif action_name == "purge":
222 count = await resource._data_source.bulk_delete(form_ids)
223 message = f"Purged {count} item(s)"
224 elif action_name == "restore":
225 count = 0
226 for item_id in form_ids:
227 updated = await resource._data_source.update(
228 item_id, {"deleted_at": None}
229 )
230 if updated is not None:
231 count += 1
232 message = f"Restored {count} item(s)"
233 else:
234 return HTMLResponse(f"Unknown action: {action_name}", status_code=400)
236 if is_htmx:
237 response = HTMLResponse(f"<p>{message}</p>")
238 response.headers["HX-Trigger"] = (
239 '{"refresh-list":true,"show-toast":{"message":"'
240 + message.replace('"', '\\"')
241 + '","type":"success"}}'
242 )
243 return response
244 resource_prefix = request.scope.get(
245 "admin_resource_prefix", resource.name or ""
246 )
247 return RedirectResponse(url=f"/admin/{resource_prefix}", status_code=302)
250class DefaultActionHandler:
251 def can_handle(self, action: str) -> bool:
252 return True
254 async def handle(
255 self, request: StarletteRequest, resource: Any, **kwargs: Any
256 ) -> Any:
257 return HTMLResponse("<h1>Unknown action</h1>")
260class ActionHandlerRegistry:
261 """Registry for action handlers."""
263 def __init__(self, config: AdminConfig, name: str, resources: dict | None = None):
264 self._config = config
265 self.name = name
266 self._resources = resources or {}
267 self._handlers: list[ResourceActionHandler] = []
268 self._initialize_handlers()
270 def _initialize_handlers(self) -> None:
271 from lexigram.admin.engine.renderer import AdminRenderer as EngineAdminRenderer
272 from lexigram.admin.resources.detail_renderer import DetailRenderer
273 from lexigram.admin.resources.form_renderer import FormRenderer
274 from lexigram.admin.resources.list_renderer import ListRenderer
276 # AdminRenderer is stateless — nav is resolved from request.app.state at
277 # render time, so a fresh instance per handler registry is safe.
278 renderer = EngineAdminRenderer()
280 list_renderer = ListRenderer(self._config, self.name, renderer)
281 detail_renderer = DetailRenderer(self._config, self.name, renderer)
282 form_renderer = FormRenderer(self._config, self.name, renderer)
284 self._handlers = [
285 ListActionHandler(list_renderer),
286 DetailActionHandler(detail_renderer),
287 CreateActionHandler(form_renderer),
288 EditActionHandler(form_renderer),
289 CloneActionHandler(),
290 RestoreActionHandler(),
291 PurgeActionHandler(),
292 DeleteActionHandler(),
293 ImportActionHandler(),
294 UserPermissionsActionHandler(self._config),
295 BulkActionHandler(),
296 DefaultActionHandler(),
297 ]
299 async def handle(
300 self, request: StarletteRequest, resource: Any, action: str
301 ) -> Any:
302 for handler in self._handlers:
303 if handler.can_handle(action):
304 return await handler.handle(request, resource)
305 return HTMLResponse("<h1>Unknown action</h1>")
308@inject
309class ResourceHandler:
310 """Handler for resource routes."""
312 def __init__(
313 self,
314 config: AdminConfig,
315 name: str,
316 action: str,
317 resources: dict | None = None,
318 ):
319 self._config = config
320 self.name = name
321 self.action = action
322 self._resources = resources or {}
323 self._registry = ActionHandlerRegistry(config, name, resources=self._resources)
325 async def __call__(self, scope: Any, receive: Any, send: Any) -> Any:
326 request = StarletteRequest(scope, receive, send)
327 scope["admin_resource_prefix"] = self.name
328 scope["admin_action"] = self.action
329 resource = self._resources.get(self.name) if self._resources else None
330 response = await self._registry.handle(request, resource, self.action)
331 await response(scope, receive, send)