Coverage for src/lexigram/admin/actions/relation.py: 96%
77 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Relation actions for relation managers (associate, attach, detach, dissociate).
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"""
10from __future__ import annotations
12from typing import Any
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
21class _RelationAction(RowAction):
22 """Base class for relation actions.
24 Resolves the target relation manager from the constructor or from
25 ``ctx.metadata["relation_manager"]``.
26 """
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
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)
53class AssociateAction(_RelationAction):
54 """Associate an existing related record with the parent.
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 """
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
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()
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 )
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 )
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 )
119class AttachAction(AssociateAction):
120 """Attach an existing related record to the parent with optional pivot data.
122 Executes the same pivot attach as :class:`AssociateAction` under a
123 distinct name/label.
124 """
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 )
145class DetachAction(_RelationAction):
146 """Detach a related record from the parent (removes pivot rows)."""
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
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()
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 )
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 )
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 )
203class DissociateAction(_RelationAction):
204 """Remove the relation to a record without deleting the record itself.
206 Works with any relation manager exposing a ``detach`` operation;
207 fails with an :class:`ActionError` when the manager has none.
208 """
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
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()
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 )
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 )
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 )
266__all__ = [
267 "AssociateAction",
268 "AttachAction",
269 "DetachAction",
270 "DissociateAction",
271]