Coverage for src/lexigram/admin/actions/relation.py: 0%

1"""Relation actions for relation managers (associate, attach, detach, dissociate). 

2 

3These actions drive relation-manager operations from the action layer, 

4mirroring Filament's relation-scoped actions. Pivot-based operations 

5(associate/attach/detach) require a :class:`BelongsToManyRelationManager` 

6configured with a pivot table and an attached data source; they fail 

7with an :class:`ActionError` otherwise. 

8""" 

9 

10from __future__ import annotations 

11 

12from typing import Any 

13 

14from lexigram.admin.actions.base import RowAction 

15from lexigram.admin.actions.exceptions import ActionError 

16from lexigram.admin.actions.types import ActionColor, ActionContext 

17from lexigram.admin.relations.belongs_to_many import BelongsToManyRelationManager 

18from lexigram.result import Err, Ok, Result 

19 

20 

21class _RelationAction(RowAction): 

22 """Base class for relation actions. 

23 

24 Resolves the target relation manager from the constructor or from 

25 ``ctx.metadata["relation_manager"]``. 

26 """ 

27 

28 def __init__( 

29 self, 

30 name: str, 

31 label: str, 

32 icon: str, 

33 color: ActionColor, 

34 relation_manager: Any = None, 

35 **kwargs: Any, 

36 ) -> None: 

37 super().__init__(name=name, label=label, icon=icon, color=color, **kwargs) 

38 self._relation_manager = relation_manager 

39 

40 def _resolve_manager(self, ctx: ActionContext) -> Result[Any, ActionError]: 

41 """Resolve the relation manager for this action.""" 

42 manager = self._relation_manager or ctx.metadata.get("relation_manager") 

43 if manager is None: 

44 return Err( 

45 ActionError( 

46 "Relation action requires a relation manager; inject one " 

47 "or set ctx.metadata['relation_manager']." 

48 ) 

49 ) 

50 return Ok(manager) 

51 

52 

53class AssociateAction(_RelationAction): 

54 """Associate an existing related record with the parent. 

55 

56 Attaches a related record through the relation manager's pivot 

57 store. The related record ID comes from ``related_id``, from 

58 ``ctx.metadata["related_id"]``, or from the record's ``id``. 

59 """ 

60 

61 def __init__( 

62 self, 

63 name: str = "associate", 

64 label: str | None = None, 

65 relation_manager: Any = None, 

66 related_id: str | None = None, 

67 pivot_data: dict[str, Any] | None = None, 

68 **kwargs: Any, 

69 ) -> None: 

70 super().__init__( 

71 name=name, 

72 label=label or "Associate", 

73 icon="link", 

74 color=ActionColor.PRIMARY, 

75 relation_manager=relation_manager, 

76 **kwargs, 

77 ) 

78 self._related_id = related_id 

79 self._pivot_data = pivot_data 

80 

81 async def execute( 

82 self, record: Any, ctx: ActionContext 

83 ) -> Result[Any, ActionError]: 

84 resolved = self._resolve_manager(ctx) 

85 if resolved.is_err(): 

86 return Err(resolved.unwrap_err()) 

87 manager = resolved.unwrap() 

88 

89 related_id = self._related_id or ctx.metadata.get("related_id") 

90 if related_id is None and isinstance(record, dict): 

91 related_id = record.get("id") 

92 if related_id is None: 

93 return Err( 

94 ActionError( 

95 "AssociateAction requires a related_id; pass one to the " 

96 "action or set ctx.metadata['related_id']." 

97 ) 

98 ) 

99 

100 if not isinstance(manager, BelongsToManyRelationManager): 

101 return Err( 

102 ActionError( 

103 f"Relation manager {type(manager).__name__} does not support " 

104 "associate; use a BelongsToManyRelationManager." 

105 ) 

106 ) 

107 

108 pivot_data = self._pivot_data or ctx.metadata.get("pivot_data") 

109 await manager.attach(related_id, pivot_data) 

110 return Ok( 

111 { 

112 "message": f"Associated {related_id}", 

113 "related_id": related_id, 

114 "action": "associate", 

115 } 

116 ) 

117 

118 

119class AttachAction(AssociateAction): 

120 """Attach an existing related record to the parent with optional pivot data. 

121 

122 Executes the same pivot attach as :class:`AssociateAction` under a 

123 distinct name/label. 

124 """ 

125 

126 def __init__( 

127 self, 

128 name: str = "attach", 

129 label: str | None = None, 

130 relation_manager: Any = None, 

131 related_id: str | None = None, 

132 pivot_data: dict[str, Any] | None = None, 

133 **kwargs: Any, 

134 ) -> None: 

135 super().__init__( 

136 name=name, 

137 label=label or "Attach", 

138 relation_manager=relation_manager, 

139 related_id=related_id, 

140 pivot_data=pivot_data, 

141 **kwargs, 

142 ) 

143 

144 

145class DetachAction(_RelationAction): 

146 """Detach a related record from the parent (removes pivot rows).""" 

147 

148 def __init__( 

149 self, 

150 name: str = "detach", 

151 label: str | None = None, 

152 relation_manager: Any = None, 

153 related_id: str | None = None, 

154 **kwargs: Any, 

155 ) -> None: 

156 super().__init__( 

157 name=name, 

158 label=label or "Detach", 

159 icon="unlink", 

160 color=ActionColor.WARNING, 

161 relation_manager=relation_manager, 

162 **kwargs, 

163 ) 

164 self._related_id = related_id 

165 

166 async def execute( 

167 self, record: Any, ctx: ActionContext 

168 ) -> Result[Any, ActionError]: 

169 resolved = self._resolve_manager(ctx) 

170 if resolved.is_err(): 

171 return Err(resolved.unwrap_err()) 

172 manager = resolved.unwrap() 

173 

174 related_id = self._related_id or ctx.metadata.get("related_id") 

175 if related_id is None and isinstance(record, dict): 

176 related_id = record.get("id") 

177 if related_id is None: 

178 return Err( 

179 ActionError( 

180 "DetachAction requires a related_id; pass one to the " 

181 "action or set ctx.metadata['related_id']." 

182 ) 

183 ) 

184 

185 if not isinstance(manager, BelongsToManyRelationManager): 

186 return Err( 

187 ActionError( 

188 f"Relation manager {type(manager).__name__} does not support " 

189 "detach; use a BelongsToManyRelationManager." 

190 ) 

191 ) 

192 

193 await manager.detach(related_id) 

194 return Ok( 

195 { 

196 "message": f"Detached {related_id}", 

197 "related_id": related_id, 

198 "action": "detach", 

199 } 

200 ) 

201 

202 

203class DissociateAction(_RelationAction): 

204 """Remove the relation to a record without deleting the record itself. 

205 

206 Works with any relation manager exposing a ``detach`` operation; 

207 fails with an :class:`ActionError` when the manager has none. 

208 """ 

209 

210 def __init__( 

211 self, 

212 name: str = "dissociate", 

213 label: str | None = None, 

214 relation_manager: Any = None, 

215 related_id: str | None = None, 

216 **kwargs: Any, 

217 ) -> None: 

218 super().__init__( 

219 name=name, 

220 label=label or "Dissociate", 

221 icon="unlink", 

222 color=ActionColor.WARNING, 

223 relation_manager=relation_manager, 

224 **kwargs, 

225 ) 

226 self._related_id = related_id 

227 

228 async def execute( 

229 self, record: Any, ctx: ActionContext 

230 ) -> Result[Any, ActionError]: 

231 resolved = self._resolve_manager(ctx) 

232 if resolved.is_err(): 

233 return Err(resolved.unwrap_err()) 

234 manager = resolved.unwrap() 

235 

236 related_id = self._related_id or ctx.metadata.get("related_id") 

237 if related_id is None and isinstance(record, dict): 

238 related_id = record.get("id") 

239 if related_id is None: 

240 return Err( 

241 ActionError( 

242 "DissociateAction requires a related_id; pass one to the " 

243 "action or set ctx.metadata['related_id']." 

244 ) 

245 ) 

246 

247 detach = getattr(manager, "detach", None) 

248 if detach is None: 

249 return Err( 

250 ActionError( 

251 f"Relation manager {type(manager).__name__} does not support " 

252 "dissociate; no detach operation available." 

253 ) 

254 ) 

255 

256 await detach(related_id) 

257 return Ok( 

258 { 

259 "message": f"Dissociated {related_id}", 

260 "related_id": related_id, 

261 "action": "dissociate", 

262 } 

263 ) 

264 

265 

266__all__ = [ 

267 "AssociateAction", 

268 "AttachAction", 

269 "DetachAction", 

270 "DissociateAction", 

271]