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

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 

7 

8from starlette.requests import Request 

9from starlette.responses import HTMLResponse, RedirectResponse, Response 

10 

11from lexigram.admin.controllers.resource.meta import T 

12from lexigram.admin.data.data_source import IDataSource as DataSourceProtocol 

13from lexigram.admin.exceptions import AdminValidationError, NotFoundError 

14from lexigram.admin.state.context import AdminContext, AdminContextManager 

15 

16if TYPE_CHECKING: 

17 from lexigram.admin.controllers.resource import ResourceController 

18 

19 

20 

21class ResourceMutationMixin: 

22 """Create, update, delete, restore.""" 

23 

24 async def create_form(self: ResourceController, request: Request) -> Response: 

25 """Show create form.""" 

26 async with AdminContextManager(request) as ctx: 

27 if ctx.is_htmx: 

28 return HTMLResponse(self.render_form_partial(ctx, None)) 

29 return HTMLResponse(self.render_form(ctx, None)) 

30 

31 async def create(self: ResourceController, request: Request) -> Response: 

32 """Create new resource.""" 

33 async with AdminContextManager(request) as ctx: 

34 form_data = request.scope.get("admin_form_data") 

35 if form_data is None: 

36 form_data = await request.form() 

37 data = dict(form_data) 

38 

39 # Validate 

40 try: 

41 validated = self.validate_create(data) 

42 except AdminValidationError as e: 

43 # Return form with errors 

44 if ctx.is_htmx: 

45 return HTMLResponse( 

46 self.render_form_partial(ctx, None, data, e.details or {}), 

47 status_code=422, 

48 ) 

49 return HTMLResponse( 

50 self.render_form(ctx, None, data, e.details or {}), 

51 status_code=422, 

52 ) 

53 

54 # Create 

55 data_source = self.get_data_source() 

56 try: 

57 created = await data_source.create(validated) 

58 except ValueError as e: 

59 return HTMLResponse(str(e), status_code=400) 

60 created_id = str(getattr(created, "id", "") if created else "") 

61 await self._emit_audit( 

62 request, 

63 f"{getattr(self.meta, 'name', 'resource')}.create", 

64 item_id=created_id, 

65 new_values=validated, 

66 ) 

67 await self._record_revision( 

68 request, created_id, validated, comment="create" 

69 ) 

70 

71 # Redirect or return success 

72 ctx.add_flash("Created successfully", "success") 

73 if ctx.is_htmx: 

74 response = Response(status_code=200) 

75 response.headers["HX-Redirect"] = f"{self.meta.prefix}/{self.meta.name}" 

76 return response 

77 

78 return RedirectResponse( 

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

80 status_code=302, 

81 ) 

82 

83 def validate_create(self: ResourceController, data: dict[str, Any]) -> dict[str, Any]: 

84 """Validate data for create. Override to customize.""" 

85 return data 

86 async def edit_form(self: ResourceController, request: Request) -> Response: 

87 """Show edit form.""" 

88 async with AdminContextManager(request) as ctx: 

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

90 

91 data_source = self.get_data_source() 

92 item = await data_source.find_one(item_id) 

93 

94 if item is None: 

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

96 

97 if ctx.is_htmx: 

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

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

100 

101 async def update(self: ResourceController, request: Request) -> Response: 

102 """Update existing resource.""" 

103 async with AdminContextManager(request) as ctx: 

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

105 form_data = request.scope.get("admin_form_data") 

106 if form_data is None: 

107 form_data = await request.form() 

108 data = dict(form_data) 

109 

110 # Validate 

111 try: 

112 validated = self.validate_update(item_id, data) 

113 except AdminValidationError as e: 

114 # Get current item for form 

115 item = await self.get_data_source().find_one(item_id) 

116 if ctx.is_htmx: 

117 return HTMLResponse( 

118 self.render_form_partial(ctx, item, data, e.details or {}), 

119 status_code=422, 

120 ) 

121 return HTMLResponse( 

122 self.render_form(ctx, item, data, e.details or {}), 

123 status_code=422, 

124 ) 

125 

126 # Update 

127 data_source = self.get_data_source() 

128 item = await data_source.find_one(item_id) 

129 can_update = getattr(self, "can_update", None) 

130 if can_update and not can_update(item): 

131 return HTMLResponse("This record cannot be updated", status_code=403) 

132 try: 

133 item = await data_source.update(item_id, validated) 

134 except ValueError as e: 

135 return HTMLResponse(str(e), status_code=400) 

136 await self._emit_audit( 

137 request, 

138 f"{getattr(self.meta, 'name', 'resource')}.update", 

139 item_id=str(item_id), 

140 new_values=validated, 

141 ) 

142 await self._record_revision( 

143 request, str(item_id), validated, comment="update" 

144 ) 

145 

146 ctx.add_flash("Updated successfully", "success") 

147 if ctx.is_htmx: 

148 response = Response(status_code=200) 

149 response.headers["HX-Redirect"] = ( 

150 f"{self.meta.prefix}/{self.meta.name}/{item_id}" 

151 ) 

152 return response 

153 

154 return RedirectResponse( 

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

156 status_code=302, 

157 ) 

158 

159 def validate_update(self: ResourceController, item_id: Any, data: dict[str, Any]) -> dict[str, Any]: 

160 """Validate data for update. Override to customize.""" 

161 return data 

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

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

164 

165 Called via HTMX GET from the Delete row action. Returns an 

166 AdminSlideOver fragment that lets the user confirm or cancel 

167 the deletion without using the native browser confirm dialog. 

168 """ 

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

170 label = self.meta.label 

171 

172 # Attempt to fetch a human-readable label for the record 

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

174 try: 

175 data_source = self.get_data_source() 

176 item = await data_source.find_one(item_id) 

177 if item: 

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

179 val = getattr(item, field, None) 

180 if val: 

181 record_label = str(val) 

182 break 

183 except Exception: # noqa: BLE001, S110 

184 pass 

185 

186 delete_url = f"{self.meta.prefix}/{self.meta.name}/{item_id}" 

187 html = render_delete_confirm( 

188 record_label=record_label, 

189 delete_url=delete_url, 

190 ) 

191 return HTMLResponse(html) 

192 async def delete(self: ResourceController, request: Request) -> Response: 

193 """Delete resource (soft or hard depending on soft_delete_enabled).""" 

194 async with AdminContextManager(request) as ctx: 

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

196 data_source = self.get_data_source() 

197 

198 if self.soft_delete_enabled: 

199 # Soft delete — stamp deleted_at instead of removing the row 

200 updated = await data_source.update( 

201 item_id, {"deleted_at": datetime.now(UTC).isoformat()} 

202 ) 

203 if updated is None: 

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

205 await self._emit_audit( 

206 request, 

207 f"{getattr(self.meta, 'name', 'resource')}.soft_delete", 

208 item_id=str(item_id), 

209 ) 

210 else: 

211 success = await data_source.delete(item_id) 

212 if not success: 

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

214 await self._emit_audit( 

215 request, 

216 f"{getattr(self.meta, 'name', 'resource')}.delete", 

217 item_id=str(item_id), 

218 ) 

219 

220 ctx.add_flash("Deleted successfully", "success") 

221 if ctx.is_htmx: 

222 response = Response(status_code=200) 

223 response.headers["HX-Redirect"] = f"{self.meta.prefix}/{self.meta.name}" 

224 return response 

225 

226 return RedirectResponse( 

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

228 status_code=302, 

229 ) 

230 

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

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

233 

234 Only available when soft_delete_enabled is True. 

235 """ 

236 async with AdminContextManager(request) as ctx: 

237 if not self.soft_delete_enabled: 

238 return HTMLResponse( 

239 "Soft delete is not enabled for this resource", status_code=400 

240 ) 

241 

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

243 data_source = self.get_data_source() 

244 

245 updated = await data_source.update(item_id, {"deleted_at": None}) 

246 if updated is None: 

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

248 await self._emit_audit( 

249 request, 

250 f"{getattr(self.meta, 'name', 'resource')}.restore", 

251 item_id=str(item_id), 

252 ) 

253 

254 ctx.add_flash("Restored successfully", "success") 

255 if ctx.is_htmx: 

256 response = Response(status_code=200) 

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

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

259 ) 

260 return response 

261 

262 return RedirectResponse( 

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

264 status_code=302, 

265 )