Coverage for src/lexigram/admin/controllers/resource/mutation.py: 0%
169 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""Create/update/delete mutations for the resource controller."""
3from __future__ import annotations
5from datetime import UTC, datetime
6from typing import TYPE_CHECKING, Any, ClassVar, get_args
8from starlette.requests import Request
9from starlette.responses import HTMLResponse, RedirectResponse, Response
11from lexigram.admin.exceptions import AdminValidationError, NotFoundError
12from lexigram.admin.state.context import AdminContextManager
14if TYPE_CHECKING:
15 from lexigram.admin.controllers.resource.meta import ResourceMeta
18class ResourceMutationMixin:
19 """Create, update, delete, restore."""
21 # Host attributes provided by sibling mixins on ResourceController.
22 meta: ResourceMeta
23 soft_delete_enabled: bool
25 render_form: Any
26 render_form_partial: Any
27 get_data_source: Any
28 _emit_audit: Any
29 _record_revision: Any
31 async def create_form(self, request: Request) -> Response:
32 """Show create form."""
33 async with AdminContextManager(request) as ctx:
34 if ctx.is_htmx:
35 return HTMLResponse(self.render_form_partial(ctx, None))
36 return HTMLResponse(self.render_form(ctx, None))
38 async def create(self, request: Request) -> Response:
39 """Create new resource."""
40 async with AdminContextManager(request) as ctx:
41 form_data = request.scope.get("admin_form_data")
42 if form_data is None:
43 form_data = await request.form()
44 data = dict(form_data)
46 # Validate
47 try:
48 validated = self.validate_create(data)
49 except AdminValidationError as e:
50 # Return form with errors
51 if ctx.is_htmx:
52 return HTMLResponse(
53 self.render_form_partial(ctx, None, data, e.details or {}),
54 status_code=422,
55 )
56 return HTMLResponse(
57 self.render_form(ctx, None, data, e.details or {}),
58 status_code=422,
59 )
61 # Create
62 data_source = self.get_data_source()
63 try:
64 created = await data_source.create(validated)
65 except ValueError as e:
66 return HTMLResponse(str(e), status_code=400)
67 created_id = str(getattr(created, "id", "") if created else "")
68 await self._emit_audit(
69 request,
70 f"{getattr(self.meta, 'name', 'resource')}.create",
71 item_id=created_id,
72 new_values=validated,
73 )
74 await self._record_revision(
75 request, created_id, validated, comment="create"
76 )
78 # Redirect or return success
79 ctx.add_flash("Created successfully", "success")
80 if ctx.is_htmx:
81 response = Response(status_code=200)
82 response.headers["HX-Redirect"] = f"{self.meta.prefix}/{self.meta.name}"
83 return response
85 return RedirectResponse(
86 url=f"{self.meta.prefix}/{self.meta.name}",
87 status_code=302,
88 )
90 _PROTECTED_FIELDS: ClassVar[frozenset[str]] = frozenset(
91 {"id", "tenant_id", "created_at", "updated_at"}
92 )
94 @classmethod
95 def _model_type(cls) -> type | None:
96 """Extract the concrete model bound via ``ResourceController[Model]``."""
97 for klass in cls.__mro__:
98 for base in getattr(klass, "__orig_bases__", ()):
99 args = get_args(base)
100 if args and isinstance(args[0], type):
101 return args[0]
102 return None
104 @classmethod
105 def _validated_model_fields(
106 cls, data: dict[str, Any]
107 ) -> dict[str, Any]:
108 """Coerce HTML form strings against the model and drop unknown keys.
110 Controllers without a bound model keep the historical raw
111 pass-through so untyped deployments are unaffected.
112 """
113 from lexigram.admin.resources.form_coercion import _coerce_form_data
115 model = cls._model_type()
116 cleaned = dict(_coerce_form_data(data, model)) if model else dict(data)
117 if not model:
118 return cleaned
119 fields = getattr(model, "model_fields", None) or getattr(
120 model, "__annotations__", {}
121 )
122 allowed = set()
123 for k, ann in fields.items():
124 if k.startswith("_") or str(ann).startswith("ClassVar"):
125 continue
126 allowed.add(k)
127 allowed -= cls._PROTECTED_FIELDS
128 return {k: v for k, v in cleaned.items() if k in allowed}
130 def validate_create(
131 self, data: dict[str, Any]
132 ) -> dict[str, Any]:
133 """Coerce form values and keep only declared model fields.
135 Unknown keys (potential mass-assignment vectors such as ``role`` or
136 ``tenant_id``) and protected columns are stripped. Override to
137 customize beyond this baseline.
138 """
139 return self._validated_model_fields(data)
141 async def edit_form(self, request: Request) -> Response:
142 """Show edit form."""
143 async with AdminContextManager(request) as ctx:
144 item_id = request.path_params.get("id")
146 data_source = self.get_data_source()
147 item = await data_source.find_one(item_id)
149 if item is None:
150 raise NotFoundError(message=f"{self.meta.label} not found")
152 if ctx.is_htmx:
153 return HTMLResponse(self.render_form_partial(ctx, item))
154 return HTMLResponse(self.render_form(ctx, item))
156 async def update(self, request: Request) -> Response:
157 """Update existing resource."""
158 async with AdminContextManager(request) as ctx:
159 item_id = request.path_params.get("id")
160 form_data = request.scope.get("admin_form_data")
161 if form_data is None:
162 form_data = await request.form()
163 data = dict(form_data)
165 # Validate
166 try:
167 validated = self.validate_update(item_id, data)
168 except AdminValidationError as e:
169 # Get current item for form
170 item = await self.get_data_source().find_one(item_id)
171 if ctx.is_htmx:
172 return HTMLResponse(
173 self.render_form_partial(ctx, item, data, e.details or {}),
174 status_code=422,
175 )
176 return HTMLResponse(
177 self.render_form(ctx, item, data, e.details or {}),
178 status_code=422,
179 )
181 # Update
182 data_source = self.get_data_source()
183 item = await data_source.find_one(item_id)
184 can_update = getattr(self, "can_update", None)
185 if can_update and not can_update(item):
186 return HTMLResponse("This record cannot be updated", status_code=403)
187 try:
188 item = await data_source.update(item_id, validated)
189 except ValueError as e:
190 return HTMLResponse(str(e), status_code=400)
191 await self._emit_audit(
192 request,
193 f"{getattr(self.meta, 'name', 'resource')}.update",
194 item_id=str(item_id),
195 new_values=validated,
196 )
197 await self._record_revision(
198 request, str(item_id), validated, comment="update"
199 )
201 ctx.add_flash("Updated successfully", "success")
202 if ctx.is_htmx:
203 response = Response(status_code=200)
204 response.headers["HX-Redirect"] = (
205 f"{self.meta.prefix}/{self.meta.name}/{item_id}"
206 )
207 return response
209 return RedirectResponse(
210 url=f"{self.meta.prefix}/{self.meta.name}/{item_id}",
211 status_code=302,
212 )
214 def validate_update(
215 self, item_id: Any, data: dict[str, Any]
216 ) -> dict[str, Any]:
217 """Coerce form values and keep only declared model fields."""
218 return self._validated_model_fields(data)
220 async def delete_confirm(self, request: Request) -> Response:
221 """Render a delete confirmation slide-over panel.
223 Called via HTMX GET from the Delete row action. Returns an
224 AdminSlideOver fragment that lets the user confirm or cancel
225 the deletion without using the native browser confirm dialog.
226 """
227 item_id = request.path_params.get("id")
228 label = self.meta.label
230 # Attempt to fetch a human-readable label for the record
231 record_label = f"{label} #{item_id}"
232 try:
233 data_source = self.get_data_source()
234 item = await data_source.find_one(item_id)
235 if item:
236 for field in ("name", "title", "email", "username", "label"):
237 val = getattr(item, field, None)
238 if val:
239 record_label = str(val)
240 break
241 except Exception: # noqa: BLE001, S110
242 pass
244 delete_url = f"{self.meta.prefix}/{self.meta.name}/{item_id}"
245 from lexigram.admin.ui.organisms.admin_slide_over import render_delete_confirm
247 html = render_delete_confirm(
248 record_label=record_label,
249 delete_url=delete_url,
250 )
251 return HTMLResponse(html)
253 async def delete(self, request: Request) -> Response:
254 """Delete resource (soft or hard depending on soft_delete_enabled)."""
255 async with AdminContextManager(request) as ctx:
256 item_id = request.path_params.get("id")
257 data_source = self.get_data_source()
259 if self.soft_delete_enabled:
260 # Soft delete — stamp deleted_at instead of removing the row
261 updated = await data_source.update(
262 item_id, {"deleted_at": datetime.now(UTC).isoformat()}
263 )
264 if updated is None:
265 raise NotFoundError(message=f"{self.meta.label} not found")
266 await self._emit_audit(
267 request,
268 f"{getattr(self.meta, 'name', 'resource')}.soft_delete",
269 item_id=str(item_id),
270 )
271 else:
272 success = await data_source.delete(item_id)
273 if not success:
274 raise NotFoundError(message=f"{self.meta.label} not found")
275 await self._emit_audit(
276 request,
277 f"{getattr(self.meta, 'name', 'resource')}.delete",
278 item_id=str(item_id),
279 )
281 ctx.add_flash("Deleted successfully", "success")
282 if ctx.is_htmx:
283 response = Response(status_code=200)
284 response.headers["HX-Redirect"] = f"{self.meta.prefix}/{self.meta.name}"
285 return response
287 return RedirectResponse(
288 url=f"{self.meta.prefix}/{self.meta.name}",
289 status_code=302,
290 )
292 async def restore(self, request: Request) -> Response:
293 """Restore a soft-deleted resource (clears deleted_at).
295 Only available when soft_delete_enabled is True.
296 """
297 async with AdminContextManager(request) as ctx:
298 if not self.soft_delete_enabled:
299 return HTMLResponse(
300 "Soft delete is not enabled for this resource", status_code=400
301 )
303 item_id = request.path_params.get("id")
304 data_source = self.get_data_source()
306 updated = await data_source.update(item_id, {"deleted_at": None})
307 if updated is None:
308 raise NotFoundError(message=f"{self.meta.label} not found")
309 await self._emit_audit(
310 request,
311 f"{getattr(self.meta, 'name', 'resource')}.restore",
312 item_id=str(item_id),
313 )
315 ctx.add_flash("Restored successfully", "success")
316 if ctx.is_htmx:
317 response = Response(status_code=200)
318 response.headers["HX-Trigger"] = (
319 '{"refresh-list":true,"show-toast":{"message":"Restored successfully","type":"success"}}'
320 )
321 return response
323 return RedirectResponse(
324 url=f"{self.meta.prefix}/{self.meta.name}",
325 status_code=302,
326 )