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

1"""Bulk actions for the resource controller.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7from starlette.requests import Request 

8from starlette.responses import HTMLResponse, RedirectResponse, Response 

9 

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

11from lexigram.admin.state.context import AdminContextManager 

12from lexigram.admin.ui.organisms.admin_slide_over import render_bulk_delete_confirm 

13from lexigram.ui import el, render_to_string 

14 

15 

16 

17 

18class ResourceBulkMixin: 

19 """Bulk action confirmations and execution.""" 

20 

21 # Host attributes provided by sibling mixins on ResourceController. 

22 meta: ResourceMeta 

23 

24 get_data_source: Any 

25 

26 async def bulk_delete_confirm(self, request: Request) -> Response: 

27 """Render a bulk delete confirmation slide-over panel. 

28 

29 Called via HTMX GET from a BulkAction button. Reads the selected 

30 record IDs from the query string (passed via ``hx-include`` of the 

31 checked checkboxes) and renders a slide-over confirmation panel. 

32 """ 

33 ids = request.query_params.getlist("ids") 

34 record_count = len(ids) 

35 

36 bulk_url = f"{self.meta.prefix}/{self.meta.name}/bulk" 

37 html = render_bulk_delete_confirm( 

38 record_count=record_count, 

39 bulk_url=bulk_url, 

40 ) 

41 return HTMLResponse(html) 

42 

43 async def bulk_purge_confirm(self, request: Request) -> Response: 

44 """Render a bulk purge confirmation slide-over panel. 

45 

46 Called via HTMX GET from a PurgeBulkAction button. Reads the 

47 selected record IDs from the query string and renders a 

48 slide-over confirmation panel posting ``action=purge``. 

49 """ 

50 ids = request.query_params.getlist("ids") 

51 record_count = len(ids) 

52 

53 bulk_url = f"{self.meta.prefix}/{self.meta.name}/bulk" 

54 html = render_bulk_delete_confirm( 

55 record_count=record_count, 

56 bulk_url=bulk_url, 

57 action="purge", 

58 title="Purge Records", 

59 heading="Confirm Bulk Purge", 

60 confirm_phrase="PURGE", 

61 subtitle=f"Purging {record_count} record{'s' if record_count != 1 else ''}", 

62 confirm_label="Purge", 

63 message=( 

64 f"You are about to permanently purge <strong>{record_count}</strong> " 

65 f"record{'s' if record_count != 1 else ''}. " 

66 "This action <strong>cannot be undone</strong>." 

67 ), 

68 ) 

69 return HTMLResponse(html) 

70 

71 async def bulk_restore_confirm(self, request: Request) -> Response: 

72 """Render a bulk restore confirmation slide-over panel. 

73 

74 Called via HTMX GET from a RestoreBulkAction button. Reads the 

75 selected record IDs from the query string and renders a 

76 slide-over confirmation panel posting ``action=restore``. 

77 """ 

78 ids = request.query_params.getlist("ids") 

79 record_count = len(ids) 

80 

81 bulk_url = f"{self.meta.prefix}/{self.meta.name}/bulk" 

82 html = render_bulk_delete_confirm( 

83 record_count=record_count, 

84 bulk_url=bulk_url, 

85 action="restore", 

86 title="Restore Records", 

87 heading="Confirm Bulk Restore", 

88 confirm_phrase="RESTORE", 

89 subtitle=f"Restoring {record_count} record{'s' if record_count != 1 else ''}", 

90 confirm_label="Restore", 

91 message=( 

92 f"You are about to restore <strong>{record_count}</strong> " 

93 f"soft-deleted record{'s' if record_count != 1 else ''}." 

94 ), 

95 variant="default", 

96 confirm_button_class=( 

97 "inline-flex items-center gap-2 rounded-lg px-4 py-2 text-sm font-medium " 

98 "text-white bg-success hover:bg-success/90 " 

99 "focus:outline-none focus:ring-2 focus:ring-success focus:ring-offset-2 " 

100 "transition-colors shadow-sm " 

101 "disabled:opacity-50 disabled:cursor-not-allowed" 

102 ), 

103 ) 

104 return HTMLResponse(html) 

105 async def bulk_action(self, request: Request) -> Response: 

106 """Handle bulk actions.""" 

107 async with AdminContextManager(request) as ctx: 

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

109 if form_data is None: 

110 form_data = await request.form() 

111 action = form_data.get("action") 

112 ids = form_data.getlist("ids") 

113 

114 if not action or not ids: 

115 return HTMLResponse("Missing action or ids", status_code=400) 

116 

117 result = await self.execute_bulk_action(action, ids) # type: ignore[arg-type] 

118 

119 ctx.add_flash(result, "success") 

120 if ctx.is_htmx: 

121 response = HTMLResponse(render_to_string(el("p", str(result)))) 

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

123 '{"refresh-list":true,"show-toast":{"message":"' 

124 + result.replace('"', '\\"') 

125 + '","type":"success"}}' 

126 ) 

127 return response 

128 

129 return RedirectResponse( 

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

131 status_code=302, 

132 ) 

133 

134 async def execute_bulk_action(self, action: str, ids: list[str]) -> str: 

135 """Execute bulk action. Override to add custom actions. 

136 

137 When the record count meets or exceeds the configured 

138 ``bulk_threshold`` (from ``TasksIntegrationConfig``), the action is 

139 dispatched through the tasks integration instead of running inline. 

140 """ 

141 if self._should_dispatch_via_tasks(len(ids)): 

142 return await self._dispatch_via_tasks(action, ids) 

143 

144 data_source = self.get_data_source() 

145 

146 if action == "delete": 

147 count = await data_source.bulk_delete(ids) 

148 return f"Deleted {count} items" 

149 

150 if action == "purge": 

151 count = await data_source.bulk_delete(ids) 

152 return f"Purged {count} items" 

153 

154 if action == "restore": 

155 restored = 0 

156 for item_id in ids: 

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

158 if updated is not None: 

159 restored += 1 

160 return f"Restored {restored} items" 

161 

162 return f"Unknown action: {action}" 

163 

164 def _should_dispatch_via_tasks(self, count: int) -> bool: 

165 """Check if the bulk count exceeds the tasks threshold.""" 

166 from lexigram.admin.integrations import get as get_integration 

167 

168 tasks = get_integration("TasksIntegration") 

169 if not tasks: 

170 return False 

171 if not tasks._enabled: 

172 return False 

173 return count >= tasks.threshold 

174 

175 async def _dispatch_via_tasks(self, action: str, ids: list[str]) -> str: 

176 """Dispatch a bulk action through the tasks integration.""" 

177 from lexigram.admin.integrations import get as get_integration 

178 

179 tasks = get_integration("TasksIntegration") 

180 if not tasks: 

181 return "Task system unavailable" 

182 

183 result = await tasks.dispatch( 

184 runner=action, 

185 action_name=action, 

186 record_ids=ids, 

187 ctx_summary=f"Bulk {action} of {len(ids)} records", 

188 ) 

189 return f"Scheduled bulk {action} for {len(ids)} records (task: {result.get('status', 'unknown')})"