Coverage for src / lexigram / admin / actions / base.py: 51%

1"""Action base classes for lexigram-admin. 

2 

3.. stability:: stable 

4 

5Defines the abstract Action hierarchy with RowAction, BulkAction, 

6and HeaderAction specializations. Each action is a frozen dataclass 

7with lifecycle hooks for visibility, authorization, confirmation, 

8and rendering. 

9""" 

10 

11from __future__ import annotations 

12 

13from abc import ABC, abstractmethod 

14from dataclasses import dataclass 

15from typing import Any, Generic, Literal, TypeVar 

16 

17from lexigram.admin.actions.exceptions import ActionError, PermissionDenied 

18from lexigram.admin.actions.types import ActionColor, ActionContext, ConfirmationConfig 

19from lexigram.result import Ok, Result 

20 

21R = TypeVar("R") 

22Outcome = TypeVar("Outcome") 

23 

24 

25@dataclass(frozen=True, kw_only=True) 

26class Action(ABC, Generic[R, Outcome]): 

27 """Abstract base for all admin actions. 

28 

29 Type parameters: 

30 R: The record type this action operates on. 

31 Outcome: The outcome type returned by execute(). 

32 

33 Attributes: 

34 name: Unique action identifier. 

35 label: Human-readable display text. 

36 icon: Optional icon identifier. 

37 color: Visual color variant for UI rendering. 

38 keyboard_shortcut: Optional keyboard shortcut (e.g. "Ctrl+E"). 

39 """ 

40 

41 name: str 

42 label: str | None = None 

43 icon: str | None = None 

44 color: ActionColor = ActionColor.GRAY 

45 keyboard_shortcut: str | None = None 

46 

47 @abstractmethod 

48 async def execute( 

49 self, record_or_records: R, ctx: ActionContext 

50 ) -> Result[Outcome, ActionError]: 

51 """Execute the action against the given record(s). 

52 

53 Args: 

54 record_or_records: The target record (or list of records 

55 for BulkAction). 

56 ctx: Action execution context. 

57 

58 Returns: 

59 Ok(outcome) on success, Err(action_error) on failure. 

60 """ 

61 ... 

62 

63 def visible_for(self, record: R, user: Any | None = None) -> bool: 

64 """Determine whether this action is visible for the given record. 

65 

66 Override to implement visibility logic. Defaults to True. 

67 """ 

68 return True 

69 

70 def is_visible( 

71 self, user: Any = None, resource_name: str | None = None, record: Any = None 

72 ) -> bool: 

73 """Determine whether this action is visible in the current context. 

74 

75 Legacy compatibility shim called by rendering code. Defaults to True. 

76 Subclasses like HeaderAction override this directly. 

77 """ 

78 return self.visible_for(record, user) 

79 

80 def authorize( 

81 self, record: R, user: Any | None = None 

82 ) -> Result[None, PermissionDenied]: 

83 """Check whether the user is authorized to execute this action. 

84 

85 Override to implement authorization logic. Defaults to Ok(None). 

86 """ 

87 return Ok(None) 

88 

89 def form(self) -> Any | None: 

90 """Return an optional form schema for parameter collection. 

91 

92 Override to return a form definition that the UI will render 

93 before executing the action. Returns None by default. 

94 """ 

95 return None 

96 

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

98 """Return optional confirmation dialog configuration. 

99 

100 Override to return a ConfirmationConfig that the UI will 

101 display before executing the action. Returns None by default. 

102 """ 

103 return None 

104 

105 # -- Button rendering -- 

106 

107 def _color_to_variant( 

108 self, 

109 ) -> Literal["primary", "secondary", "danger", "ghost", "link"]: 

110 """Map ActionColor to ActionButton variant string.""" 

111 mapping = { 

112 ActionColor.GRAY: "ghost", 

113 ActionColor.PRIMARY: "primary", 

114 ActionColor.SECONDARY: "secondary", 

115 ActionColor.SUCCESS: "secondary", 

116 ActionColor.WARNING: "warning", 

117 ActionColor.DANGER: "danger", 

118 ActionColor.INFO: "secondary", 

119 } 

120 return mapping[self.color] # type: ignore[return-value] 

121 

122 def _get_url(self, record: R, ctx: ActionContext) -> str | None: 

123 """Build the endpoint URL for this action. 

124 

125 Override in subclasses to provide action-specific URL patterns. 

126 Returns None if the URL cannot be determined (button hidden). 

127 """ 

128 return None 

129 

130 def get_url(self, record: Any = None) -> str | None: 

131 """Public URL accessor for legacy rendering code. 

132 

133 Delegates to _get_url with a minimal ActionContext when called 

134 from non-framework rendering paths. 

135 """ 

136 return None 

137 

138 def _get_htmx_attrs( 

139 self, url: str, record: R, ctx: ActionContext 

140 ) -> dict[str, str]: 

141 """Build HTMX attributes for the action button. 

142 

143 Override in subclasses to customize HTMX behavior. 

144 """ 

145 return { 

146 "hx-get": url, 

147 "hx-target": "#table-data", 

148 "hx-swap": "innerHTML", 

149 } 

150 

151 def render_button(self, record: R, ctx: ActionContext) -> str: 

152 """Render the action as an HTML button string. 

153 

154 Args: 

155 record: The target record context for rendering. 

156 ctx: Action context. 

157 

158 Returns: 

159 HTML string for the button element, or empty string if 

160 the action is not visible or cannot be resolved. 

161 """ 

162 if not self.visible_for(record, ctx.user): 

163 return "" 

164 

165 url = self._get_url(record, ctx) 

166 if not url: 

167 return "" 

168 

169 variant = self._color_to_variant() 

170 htmx_attrs = self._get_htmx_attrs(url, record, ctx) 

171 

172 from lexigram.ui import ActionButton 

173 

174 button = ActionButton( 

175 label=self.label or self.name, 

176 variant=variant, 

177 icon=self.icon, 

178 size="sm", 

179 **htmx_attrs, # type: ignore[arg-type] 

180 ) 

181 result = button.render() 

182 return str(result) if result else "" 

183 

184 

185class RowAction(Action[Any, Any]): 

186 """Action that operates on a single record. 

187 

188 Example use: View, Edit, Delete, Duplicate for a table row. 

189 """ 

190 

191 @staticmethod 

192 def _get_record_id(record: Any) -> str: 

193 """Extract a record identifier from dict or object.""" 

194 if record is None: 

195 return "" 

196 if isinstance(record, dict): 

197 return str(record.get("id", "")) 

198 if hasattr(record, "id"): 

199 return str(record.id) 

200 return "" 

201 

202 def _get_url(self, record: Any, ctx: ActionContext) -> str | None: 

203 record_id = self._get_record_id(record) 

204 if not record_id: 

205 return None 

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

207 return f"{prefix}/{record_id}/{self.name}" 

208 

209 

210class BulkAction(Action[list[Any], Any]): 

211 """Action that operates on multiple records at once. 

212 

213 Example use: Bulk delete, bulk status change, bulk assign. 

214 

215 When *task_runner* is set and the record count equals or exceeds the 

216 configured ``bulk_threshold`` (from ``TasksIntegrationConfig``), execution 

217 is dispatched through the tasks integration instead of running inline. 

218 """ 

219 

220 task_runner: str | None = None 

221 """Name of the task runner to use when dispatching via lexigram-tasks.""" 

222 

223 def _get_url(self, records: list[Any], ctx: ActionContext) -> str | None: 

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

225 return f"{prefix}/bulk/{self.name}" 

226 

227 

228class HeaderAction(Action[None, Any]): 

229 """Action with no record context, rendered in header areas. 

230 

231 Example use: Export all, settings, global create. 

232 """ 

233 

234 def _get_url(self, record: None, ctx: ActionContext) -> str | None: 

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

236 return f"{prefix}/{self.name}" 

237 

238 

239__all__ = [ 

240 "Action", 

241 "BulkAction", 

242 "HeaderAction", 

243 "RowAction", 

244]