Coverage for src/lexigram/admin/controllers/resource/mutation.py: 0%

1"""Create/update/delete mutations for the resource controller.""" 

2 

3from __future__ import annotations 

4 

5from datetime import UTC, datetime 

6from typing import TYPE_CHECKING, Any, ClassVar, get_args 

7 

8from starlette.requests import Request 

9from starlette.responses import HTMLResponse, RedirectResponse, Response 

10 

11from lexigram.admin.exceptions import AdminValidationError, NotFoundError 

12from lexigram.admin.state.context import AdminContextManager 

13 

14if TYPE_CHECKING: 

15 from lexigram.admin.controllers.resource.meta import ResourceMeta 

16 

17 

18class ResourceMutationMixin: 

19 """Create, update, delete, restore.""" 

20 

21 # Host attributes provided by sibling mixins on ResourceController. 

22 meta: ResourceMeta 

23 soft_delete_enabled: bool 

24 

25 render_form: Any 

26 render_form_partial: Any 

27 get_data_source: Any 

28 _emit_audit: Any 

29 _record_revision: Any 

30 

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)) 

37 

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) 

45 

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 ) 

60 

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 ) 

77 

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 

84 

85 return RedirectResponse( 

86 url=f"{self.meta.prefix}/{self.meta.name}", 

87 status_code=302, 

88 ) 

89 

90 _PROTECTED_FIELDS: ClassVar[frozenset[str]] = frozenset( 

91 {"id", "tenant_id", "created_at", "updated_at"} 

92 ) 

93 

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 

103 

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. 

109 

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 

114 

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} 

129 

130 def validate_create( 

131 self, data: dict[str, Any] 

132 ) -> dict[str, Any]: 

133 """Coerce form values and keep only declared model fields. 

134 

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) 

140 

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") 

145 

146 data_source = self.get_data_source() 

147 item = await data_source.find_one(item_id) 

148 

149 if item is None: 

150 raise NotFoundError(message=f"{self.meta.label} not found") 

151 

152 if ctx.is_htmx: 

153 return HTMLResponse(self.render_form_partial(ctx, item)) 

154 return HTMLResponse(self.render_form(ctx, item)) 

155 

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) 

164 

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 ) 

180 

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 ) 

200 

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 

208 

209 return RedirectResponse( 

210 url=f"{self.meta.prefix}/{self.meta.name}/{item_id}", 

211 status_code=302, 

212 ) 

213 

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) 

219 

220 async def delete_confirm(self, request: Request) -> Response: 

221 """Render a delete confirmation slide-over panel. 

222 

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 

229 

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 

243 

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 

246 

247 html = render_delete_confirm( 

248 record_label=record_label, 

249 delete_url=delete_url, 

250 ) 

251 return HTMLResponse(html) 

252 

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() 

258 

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 ) 

280 

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 

286 

287 return RedirectResponse( 

288 url=f"{self.meta.prefix}/{self.meta.name}", 

289 status_code=302, 

290 ) 

291 

292 async def restore(self, request: Request) -> Response: 

293 """Restore a soft-deleted resource (clears deleted_at). 

294 

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 ) 

302 

303 item_id = request.path_params.get("id") 

304 data_source = self.get_data_source() 

305 

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 ) 

314 

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 

322 

323 return RedirectResponse( 

324 url=f"{self.meta.prefix}/{self.meta.name}", 

325 status_code=302, 

326 )