Coverage for src/lexigram/admin/relations/belongs_to_many.py: 75%
166 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""BelongsToMany (many-to-many) relation manager with pivot data support."""
3from __future__ import annotations
5from typing import TYPE_CHECKING, Any
7from lexigram.admin.data.query import QuerySpec
8from lexigram.admin.relations.errors import RelationPersistenceError
9from lexigram.admin.relations.manager_ext import RelationManager
10from lexigram.serialization import dumps_str, loads_str
11from lexigram.ui import el, render_to_string
13if TYPE_CHECKING:
14 from collections.abc import Sequence
16 from starlette.requests import Request
17 from starlette.responses import Response
20class BelongsToManyRelationManager(RelationManager):
21 """Relation manager for many-to-many relationships through a pivot table.
23 Provides attach/detach/sync operations and inline pivot data
24 editing for each related record.
26 Example:
27 class UserRolesRelationManager(BelongsToManyRelationManager):
28 relationship_name = "roles"
29 pivot_table = "user_roles"
30 pivot_columns = ["assigned_at", "is_primary"]
31 related_key = "role_id"
32 related_key_local = "user_id"
34 async def get_query(self):
35 return await role_service.list()
36 """
38 pivot_table: str = ""
39 pivot_columns: list[str] = []
40 related_key: str = "related_id"
41 related_key_local: str = "parent_id"
43 @classmethod
44 def table(cls, table_config: Any = None) -> list[Any]:
45 return []
47 def _require_persistence(self) -> None:
48 """Raise unless pivot persistence is configured."""
49 if not self.pivot_table:
50 raise RelationPersistenceError(
51 "BelongsToManyRelationManager requires a pivot_table "
52 f"({self.get_relationship_name()})"
53 )
54 if self._data_source is None:
55 raise RelationPersistenceError(
56 "BelongsToManyRelationManager requires an attached data source; "
57 "pass data_source to the constructor or call set_data_source()"
58 )
60 def _row_id(self, row: Any) -> Any:
61 """Extract a row's primary key."""
62 if isinstance(row, dict):
63 return row.get("id") or row.get("pk")
64 return getattr(row, "id", None) or getattr(row, "pk", None)
66 def _row_value(self, row: Any, field: str) -> Any:
67 """Extract a field value from a record."""
68 if isinstance(row, dict):
69 return row.get(field)
70 return getattr(row, field, None)
72 async def _find_pivot_rows(self) -> list[Any]:
73 """Look up pivot rows for the current parent through the data source."""
74 query = QuerySpec().with_where_eq(self.related_key_local, self.parent_id)
75 result = await self._data_source.find_many(query)
76 if result is None:
77 return []
78 return list(result.items) if hasattr(result, "items") else []
80 async def _matching_pivot_rows(self, related_id: str) -> list[Any]:
81 """Pivot rows linking the current parent to the given related record."""
82 rows = await self._find_pivot_rows()
83 return [
84 row
85 for row in rows
86 if str(self._row_value(row, self.related_key)) == str(related_id)
87 ]
89 async def attach(
90 self, related_id: str, pivot_data: dict[str, Any] | None = None
91 ) -> None:
92 """Attach a related record with optional pivot data.
94 Persists a pivot row through the attached data source.
96 Args:
97 related_id: ID of the related record to attach.
98 pivot_data: Optional values for configured pivot columns.
100 Raises:
101 RelationPersistenceError: When no pivot table or data
102 source is configured.
103 """
104 self._require_persistence()
105 row: dict[str, Any] = {
106 self.related_key_local: self.parent_id,
107 self.related_key: related_id,
108 }
109 if pivot_data:
110 if self.pivot_columns:
111 row.update(
112 {k: v for k, v in pivot_data.items() if k in self.pivot_columns}
113 )
114 else:
115 row.update(pivot_data)
116 await self._data_source.create(row)
118 async def detach(self, related_id: str) -> None:
119 """Detach a related record by removing its pivot rows.
121 Args:
122 related_id: ID of the related record to detach.
124 Raises:
125 RelationPersistenceError: When no pivot table or data
126 source is configured.
127 """
128 self._require_persistence()
129 rows = await self._matching_pivot_rows(related_id)
130 ids = [self._row_id(row) for row in rows if self._row_id(row) is not None]
131 if ids:
132 await self._data_source.bulk_delete(ids)
134 async def sync(
135 self,
136 related_ids: Sequence[str],
137 pivot_data_map: dict[str, dict[str, Any]] | None = None,
138 ) -> None:
139 """Sync related records, detaching any not in the list and attaching new ones.
141 Args:
142 related_ids: IDs to keep attached.
143 pivot_data_map: Optional mapping of related_id -> pivot data.
145 Raises:
146 RelationPersistenceError: When no pivot table or data
147 source is configured.
148 """
149 current = await self.get_attached_ids()
150 pivot_data_map = pivot_data_map or {}
151 for related_id in current:
152 if related_id not in related_ids:
153 await self.detach(related_id)
154 for related_id in related_ids:
155 if related_id not in current:
156 await self.attach(related_id, pivot_data_map.get(related_id))
158 async def get_attached_ids(self) -> list[str]:
159 """Return IDs of currently attached related records."""
160 if self._data_source is None:
161 return []
162 rows = await self._find_pivot_rows()
163 return [
164 str(self._row_value(row, self.related_key))
165 for row in rows
166 if self._row_value(row, self.related_key) is not None
167 ]
169 async def get_pivot_data(self, related_id: str) -> dict[str, Any] | None:
170 """Return pivot data for a single attached record."""
171 if self._data_source is None:
172 return None
173 rows = await self._matching_pivot_rows(related_id)
174 if not rows:
175 return None
176 row = rows[0]
177 if self.pivot_columns:
178 return {col: self._row_value(row, col) for col in self.pivot_columns}
179 if isinstance(row, dict):
180 return dict(row)
181 return (
182 {key: getattr(row, key) for key in vars(row) if not key.startswith("_")}
183 if hasattr(row, "__dict__")
184 else None
185 )
187 async def update_pivot(self, related_id: str, pivot_data: dict[str, Any]) -> None:
188 """Update pivot data for an attached record.
190 Args:
191 related_id: ID of the attached related record.
192 pivot_data: Values for configured pivot columns.
194 Raises:
195 RelationPersistenceError: When no pivot table or data
196 source is configured.
197 """
198 self._require_persistence()
199 rows = await self._matching_pivot_rows(related_id)
200 if not rows:
201 return
202 row_id = self._row_id(rows[0])
203 if row_id is None:
204 return
205 updates = (
206 {k: v for k, v in pivot_data.items() if k in self.pivot_columns}
207 if self.pivot_columns
208 else dict(pivot_data)
209 )
210 if updates:
211 await self._data_source.update(row_id, updates)
213 async def render(self, request: Request, resource_name: str = "") -> str:
214 items = await self.get_query()
215 attached_ids = await self.get_attached_ids()
216 rel_name = self.get_relationship_name()
218 rows: list[Any] = []
219 for item in items:
220 item_id = str(getattr(item, "id", ""))
221 is_attached = item_id in attached_ids
222 label = str(getattr(item, "name", item_id))
224 pivot_data = await self.get_pivot_data(item_id) if is_attached else None
225 rows.append(
226 self._build_row(
227 resource_name,
228 item_id,
229 label,
230 is_attached,
231 self._render_pivot_cells(item_id, pivot_data),
232 )
233 )
235 header = el(
236 "div",
237 el(
238 "h3",
239 rel_name.replace("_", " ").title(),
240 class_="text-lg font-medium text-foreground",
241 ),
242 el(
243 "div",
244 el(
245 "input",
246 type="text",
247 class_="px-3 py-1.5 text-sm border rounded-lg",
248 placeholder="Search...",
249 id=f"search-{rel_name}",
250 hx_trigger="keyup changed delay:300ms",
251 hx_get=f"/admin/{resource_name}/{self.parent_id}/relations/{rel_name}",
252 hx_target=f"#relation-panel-{rel_name}",
253 hx_select=".relation-panel",
254 ),
255 class_="flex gap-2",
256 ),
257 class_="flex items-center justify-between mb-4",
258 )
260 table = el(
261 "table",
262 el(
263 "thead",
264 el(
265 "tr",
266 el(
267 "th",
268 "Attach",
269 class_="px-4 py-2 text-left text-xs font-medium text-muted-foreground uppercase",
270 ),
271 el(
272 "th",
273 "Record",
274 class_="px-4 py-2 text-left text-xs font-medium text-muted-foreground uppercase",
275 ),
276 *self._pivot_header_elements(),
277 el(
278 "th",
279 "ID",
280 class_="px-4 py-2 text-left text-xs font-medium text-muted-foreground uppercase",
281 ),
282 ),
283 class_="bg-muted dark:bg-card",
284 ),
285 el("tbody", *rows, class_="divide-y divide-border"),
286 class_="min-w-full divide-y divide-border",
287 )
289 return render_to_string(
290 el(
291 "div",
292 header,
293 table,
294 el(
295 "button",
296 "Save",
297 type="button",
298 class_="px-3 py-1.5 text-sm font-medium text-white bg-primary-600 rounded-lg hover:bg-primary-700",
299 hx_post=f"/admin/{resource_name}/{self.parent_id}/relations/{rel_name}/sync",
300 hx_target=f"#relation-panel-{rel_name}",
301 hx_swap="outerHTML",
302 ),
303 class_="relation-panel p-4",
304 id=f"relation-panel-{rel_name}",
305 )
306 )
308 def _pivot_header_elements(self) -> list[Any]:
309 """Return table header cell elements for the pivot columns."""
310 return [
311 el(
312 "th",
313 c.replace("_", " ").title(),
314 class_="px-4 py-2 text-left text-xs font-medium text-muted-foreground uppercase",
315 )
316 for c in self.pivot_columns
317 ]
319 def _render_pivot_headers(self) -> str:
320 """Render the pivot column table headers as HTML."""
321 return render_to_string(self._pivot_header_elements())
323 def _build_row(
324 self,
325 resource_name: str,
326 item_id: str,
327 label: str,
328 is_attached: bool,
329 pivot_cells: list[Any],
330 ) -> Any:
331 """Build a single belongs-to-many row element."""
332 rel_name = self.get_relationship_name()
333 return el(
334 "tr",
335 el(
336 "td",
337 el(
338 "input",
339 type="checkbox",
340 class_="belongs-to-many-checkbox rounded border-border text-primary-600 focus:ring-primary-500",
341 data_related_id=item_id,
342 checked="checked" if is_attached else None,
343 hx_post=f"/admin/{resource_name}/{self.parent_id}/relations/{rel_name}/toggle",
344 hx_vals=dumps_str({"related_id": item_id}),
345 hx_target="closest tr",
346 hx_swap="outerHTML",
347 ),
348 ),
349 el("td", label, class_="px-4 py-2 text-sm text-foreground"),
350 *pivot_cells,
351 el("td", item_id, class_="px-4 py-2 text-sm text-muted-foreground"),
352 class_="bg-primary-50 dark:bg-primary-900/20" if is_attached else None,
353 )
355 def _render_pivot_cells(
356 self, related_id: str, pivot_data: dict[str, Any] | None
357 ) -> list[Any]:
358 """Return pivot cell elements for a single related record."""
359 if not self.pivot_columns:
360 return []
361 cells: list[Any] = []
362 for col in self.pivot_columns:
363 value = (pivot_data or {}).get(col, "")
364 cells.append(
365 el(
366 "td",
367 el(
368 "input",
369 type="text",
370 class_="px-2 py-1 text-sm border rounded w-full",
371 value=value,
372 name=f"pivot_{col}_{related_id}",
373 hx_post=f"/admin/{self.parent_id}/relations/{self.get_relationship_name()}/pivot/{related_id}",
374 hx_trigger="change",
375 hx_swap="none",
376 ),
377 class_="px-4 py-2",
378 )
379 )
380 return cells
382 def get_pivot_routes(self, resource_name: str) -> list[Any]:
383 """Return additional routes for pivot operations."""
384 from starlette.responses import HTMLResponse
385 from starlette.routing import Route
387 prefix = f"/admin/{resource_name}/{self.parent_id}/relations/{self.get_relationship_name()}"
389 async def _handle_toggle(request: Any) -> Response:
390 if request.headers.get("content-type") == "application/json":
391 body = await request.json()
392 else:
393 body = request.scope.get("admin_form_data")
394 if body is None:
395 body = await request.form()
396 related_id = body.get("related_id", "")
397 attached = await self.get_attached_ids()
398 if related_id in attached:
399 await self.detach(related_id)
400 else:
401 await self.attach(related_id)
402 return await self._render_single_row(request, resource_name, related_id)
404 async def _handle_sync(request: Any) -> Response:
405 if request.headers.get("content-type") == "application/json":
406 body = await request.json()
407 else:
408 body = request.scope.get("admin_form_data")
409 if body is None:
410 body = await request.form()
411 raw_ids = body.get("related_ids", "")
412 if isinstance(raw_ids, str):
413 ids = (
414 loads_str(raw_ids)
415 if raw_ids.startswith("[")
416 else raw_ids.split(",")
417 )
418 else:
419 ids = raw_ids or []
420 await self.sync(ids)
421 html = await self.render(request, resource_name)
422 return HTMLResponse(html)
424 async def _handle_pivot_update(request: Any) -> Response:
425 related_id = request.path_params.get("related_id", "")
426 form = request.scope.get("admin_form_data")
427 if form is None:
428 form = await request.form()
429 pivot_data = dict(form)
430 await self.update_pivot(related_id, pivot_data)
431 return HTMLResponse("")
433 return [
434 Route(f"{prefix}/toggle", _handle_toggle, methods=["POST"]),
435 Route(f"{prefix}/sync", _handle_sync, methods=["POST"]),
436 Route(
437 f"{prefix}/pivot/{{related_id}}", _handle_pivot_update, methods=["POST"]
438 ),
439 ]
441 async def _render_single_row(
442 self, request: Any, resource_name: str, related_id: str
443 ) -> Any:
444 from starlette.responses import HTMLResponse
446 items = await self.get_query()
447 attached_ids = await self.get_attached_ids()
448 item = next((i for i in items if str(getattr(i, "id", "")) == related_id), None)
449 if item is None:
450 return HTMLResponse("")
452 is_attached = related_id in attached_ids
453 label = str(getattr(item, "name", related_id))
454 pivot_data = await self.get_pivot_data(related_id) if is_attached else None
455 return HTMLResponse(
456 render_to_string(
457 self._build_row(
458 resource_name,
459 related_id,
460 label,
461 is_attached,
462 self._render_pivot_cells(related_id, pivot_data),
463 )
464 )
465 )