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

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

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7from starlette.requests import Request 

8from starlette.responses import HTMLResponse, RedirectResponse, Response 

9 

10from lexigram.admin.state.context import AdminContextManager 

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

12from lexigram.ui import el, render_to_string 

13 

14if TYPE_CHECKING: 

15 from lexigram.admin.controllers.resource import ResourceController 

16 

17 

18 

19class ResourceBulkMixin: 

20 """Bulk action confirmations and execution.""" 

21 

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

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

24 

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

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

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

28 """ 

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

30 record_count = len(ids) 

31 

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

33 html = render_bulk_delete_confirm( 

34 record_count=record_count, 

35 bulk_url=bulk_url, 

36 ) 

37 return HTMLResponse(html) 

38 

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

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

41 

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

43 selected record IDs from the query string and renders a 

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

45 """ 

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

47 record_count = len(ids) 

48 

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

50 html = render_bulk_delete_confirm( 

51 record_count=record_count, 

52 bulk_url=bulk_url, 

53 action="purge", 

54 title="Purge Records", 

55 heading="Confirm Bulk Purge", 

56 confirm_phrase="PURGE", 

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

58 confirm_label="Purge", 

59 message=( 

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

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

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

63 ), 

64 ) 

65 return HTMLResponse(html) 

66 

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

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

69 

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

71 selected record IDs from the query string and renders a 

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

73 """ 

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

75 record_count = len(ids) 

76 

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

78 html = render_bulk_delete_confirm( 

79 record_count=record_count, 

80 bulk_url=bulk_url, 

81 action="restore", 

82 title="Restore Records", 

83 heading="Confirm Bulk Restore", 

84 confirm_phrase="RESTORE", 

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

86 confirm_label="Restore", 

87 message=( 

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

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

90 ), 

91 variant="default", 

92 confirm_button_class=( 

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

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

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

96 "transition-colors shadow-sm " 

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

98 ), 

99 ) 

100 return HTMLResponse(html) 

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

102 """Handle bulk actions.""" 

103 async with AdminContextManager(request) as ctx: 

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

105 if form_data is None: 

106 form_data = await request.form() 

107 action = form_data.get("action") 

108 ids = form_data.getlist("ids") 

109 

110 if not action or not ids: 

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

112 

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

114 

115 ctx.add_flash(result, "success") 

116 if ctx.is_htmx: 

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

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

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

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

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

122 ) 

123 return response 

124 

125 return RedirectResponse( 

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

127 status_code=302, 

128 ) 

129 

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

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

132 

133 When the record count meets or exceeds the configured 

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

135 dispatched through the tasks integration instead of running inline. 

136 """ 

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

138 return await self._dispatch_via_tasks(action, ids) 

139 

140 data_source = self.get_data_source() 

141 

142 if action == "delete": 

143 count = await data_source.bulk_delete(ids) 

144 return f"Deleted {count} items" 

145 

146 if action == "purge": 

147 count = await data_source.bulk_delete(ids) 

148 return f"Purged {count} items" 

149 

150 if action == "restore": 

151 restored = 0 

152 for item_id in ids: 

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

154 if updated is not None: 

155 restored += 1 

156 return f"Restored {restored} items" 

157 

158 return f"Unknown action: {action}" 

159 

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

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

162 from lexigram.admin.integrations import get as get_integration 

163 

164 tasks = get_integration("TasksIntegration") 

165 if not tasks: 

166 return False 

167 if not tasks._enabled: 

168 return False 

169 return count >= tasks.threshold 

170 

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

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

173 from lexigram.admin.integrations import get as get_integration 

174 

175 tasks = get_integration("TasksIntegration") 

176 if not tasks: 

177 return "Task system unavailable" 

178 

179 result = await tasks.dispatch( 

180 runner=action, 

181 action_name=action, 

182 record_ids=ids, 

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

184 ) 

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