Coverage for src/lexigram/admin/actions/standard/bulk.py: 100%

1"""Ready-to-use bulk actions (delete, purge, restore). 

2 

3Part of the ``lexigram.admin.actions.standard`` package. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import Any 

9 

10from lexigram.admin.actions.base import BulkAction 

11from lexigram.admin.actions.standard.utils import _extract_id, _resolve_data_source 

12from lexigram.admin.actions.exceptions import ActionError 

13from lexigram.admin.actions.types import ( 

14 ActionColor, 

15 ActionContext, 

16 ConfirmationConfig, 

17) 

18from lexigram.result import Err, Ok, Result 

19class DeleteBulkAction(BulkAction): 

20 """Delete multiple selected records.""" 

21 

22 def __init__( 

23 self, 

24 name: str = "delete", 

25 label: str | None = None, 

26 ) -> None: 

27 super().__init__( 

28 name=name, 

29 label=label or "Delete Selected", 

30 icon="trash", 

31 color=ActionColor.DANGER, 

32 ) 

33 

34 def _get_htmx_attrs( 

35 self, url: str, records: list[Any], ctx: ActionContext 

36 ) -> dict[str, str]: 

37 # Open bulk delete confirmation slide-over instead of native hx-confirm 

38 prefix = ctx.resource_prefix or f"/{ctx.resource_name}" 

39 confirm_url = f"{prefix}/bulk-delete-confirm" 

40 return { 

41 "hx-get": confirm_url, 

42 "hx-target": "#slide-over-container", 

43 "hx-swap": "innerHTML", 

44 "hx-push-url": "false", 

45 "hx-include": "#lexigram-table [name='ids']:checked", 

46 } 

47 

48 def confirm(self) -> ConfirmationConfig | None: 

49 return ConfirmationConfig( 

50 title="Delete Selected Records", 

51 message="Are you sure you want to delete the selected records? " 

52 "This action cannot be undone.", 

53 style=ActionColor.DANGER, 

54 ) 

55 

56 async def execute(self, records: list[Any], ctx: ActionContext) -> Result[Any, Any]: 

57 count = len(records) 

58 return Ok({"message": f"Deleted {count} record(s)", "deleted_count": count}) 

59class PurgeBulkAction(BulkAction): 

60 """Permanently delete multiple selected records. 

61 

62 Deletes are issued in chunks via the data source's ``bulk_delete``, 

63 mirroring Filament's ``chunkSelectedRecords`` behaviour for large 

64 selections. 

65 """ 

66 

67 def __init__( 

68 self, 

69 name: str = "purge", 

70 label: str | None = None, 

71 data_source: Any | None = None, 

72 chunk_size: int = 200, 

73 ) -> None: 

74 super().__init__( 

75 name=name, 

76 label=label or "Purge Selected", 

77 icon="trash-2", 

78 color=ActionColor.DANGER, 

79 ) 

80 self._data_source = data_source 

81 self._chunk_size = chunk_size 

82 

83 def _get_htmx_attrs( 

84 self, url: str, records: list[Any], ctx: ActionContext 

85 ) -> dict[str, str]: 

86 # Open bulk purge confirmation slide-over instead of native hx-confirm 

87 prefix = ctx.resource_prefix or f"/{ctx.resource_name}" 

88 confirm_url = f"{prefix}/bulk-purge-confirm" 

89 return { 

90 "hx-get": confirm_url, 

91 "hx-target": "#slide-over-container", 

92 "hx-swap": "innerHTML", 

93 "hx-push-url": "false", 

94 "hx-include": "#lexigram-table [name='ids']:checked", 

95 } 

96 

97 def confirm(self) -> ConfirmationConfig | None: 

98 return ConfirmationConfig( 

99 title="Purge Selected Records", 

100 message="Are you sure you want to permanently purge the selected " 

101 "records? This action cannot be undone.", 

102 style=ActionColor.DANGER, 

103 ) 

104 

105 async def execute(self, records: list[Any], ctx: ActionContext) -> Result[Any, Any]: 

106 data_source = _resolve_data_source(ctx, self._data_source) 

107 if data_source is None: 

108 return Err( 

109 ActionError( 

110 "Purge requires a data source; inject one or set ctx.data_source." 

111 ) 

112 ) 

113 ids = [ 

114 item_id 

115 for record in records 

116 if (item_id := _extract_id(record)) is not None 

117 ] 

118 chunk_size = self._chunk_size or len(ids) or 1 

119 purged = 0 

120 for start in range(0, len(ids), chunk_size): 

121 purged += await data_source.bulk_delete(ids[start : start + chunk_size]) 

122 return Ok({"message": f"Purged {purged} record(s)", "purged_count": purged}) 

123class RestoreBulkAction(BulkAction): 

124 """Restore multiple soft-deleted records.""" 

125 

126 def __init__( 

127 self, 

128 name: str = "restore", 

129 label: str | None = None, 

130 data_source: Any | None = None, 

131 ) -> None: 

132 super().__init__( 

133 name=name, 

134 label=label or "Restore Selected", 

135 icon="rotate-ccw", 

136 color=ActionColor.SUCCESS, 

137 ) 

138 self._data_source = data_source 

139 

140 def _get_htmx_attrs( 

141 self, url: str, records: list[Any], ctx: ActionContext 

142 ) -> dict[str, str]: 

143 # Open bulk restore confirmation slide-over instead of native hx-confirm 

144 prefix = ctx.resource_prefix or f"/{ctx.resource_name}" 

145 confirm_url = f"{prefix}/bulk-restore-confirm" 

146 return { 

147 "hx-get": confirm_url, 

148 "hx-target": "#slide-over-container", 

149 "hx-swap": "innerHTML", 

150 "hx-push-url": "false", 

151 "hx-include": "#lexigram-table [name='ids']:checked", 

152 } 

153 

154 def confirm(self) -> ConfirmationConfig | None: 

155 return ConfirmationConfig( 

156 title="Restore Selected Records", 

157 message="Are you sure you want to restore the selected records?", 

158 style=ActionColor.SUCCESS, 

159 ) 

160 

161 async def execute(self, records: list[Any], ctx: ActionContext) -> Result[Any, Any]: 

162 data_source = _resolve_data_source(ctx, self._data_source) 

163 if data_source is None: 

164 return Err( 

165 ActionError( 

166 "Restore requires a data source; inject one or set ctx.data_source." 

167 ) 

168 ) 

169 ids = [ 

170 item_id 

171 for record in records 

172 if (item_id := _extract_id(record)) is not None 

173 ] 

174 restored = 0 

175 for item_id in ids: 

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

177 if updated is not None: 

178 restored += 1 

179 return Ok( 

180 {"message": f"Restored {restored} record(s)", "restored_count": restored} 

181 )