Coverage for src / lexigram / admin / relations / belongs_to_many.py: 22%

1"""BelongsToMany (many-to-many) relation manager with pivot data support.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7from lexigram.admin.relations.manager_ext import RelationManager 

8from lexigram.serialization import loads_str 

9 

10if TYPE_CHECKING: 

11 from collections.abc import Sequence 

12 

13 from starlette.requests import Request 

14 from starlette.responses import Response 

15 

16 

17class BelongsToManyRelationManager(RelationManager): 

18 """Relation manager for many-to-many relationships through a pivot table. 

19 

20 Provides attach/detach/sync operations and inline pivot data 

21 editing for each related record. 

22 

23 Example: 

24 class UserRolesRelationManager(BelongsToManyRelationManager): 

25 relationship_name = "roles" 

26 pivot_table = "user_roles" 

27 pivot_columns = ["assigned_at", "is_primary"] 

28 related_key = "role_id" 

29 related_key_local = "user_id" 

30 

31 async def get_query(self): 

32 return await role_service.list() 

33 """ 

34 

35 pivot_table: str = "" 

36 pivot_columns: list[str] = [] 

37 related_key: str = "related_id" 

38 related_key_local: str = "parent_id" 

39 

40 @classmethod 

41 def table(cls, table_config: Any = None) -> list[Any]: 

42 return [] 

43 

44 async def attach( 

45 self, related_id: str, pivot_data: dict[str, Any] | None = None 

46 ) -> None: 

47 """Attach a related record with optional pivot data.""" 

48 

49 async def detach(self, related_id: str) -> None: 

50 """Detach a related record.""" 

51 

52 async def sync( 

53 self, 

54 related_ids: Sequence[str], 

55 pivot_data_map: dict[str, dict[str, Any]] | None = None, 

56 ) -> None: 

57 """Sync related records, detaching any not in the list and attaching new ones. 

58 

59 Args: 

60 related_ids: IDs to keep attached. 

61 pivot_data_map: Optional mapping of related_id -> pivot data. 

62 """ 

63 

64 async def get_attached_ids(self) -> list[str]: 

65 """Return IDs of currently attached related records.""" 

66 return [] 

67 

68 async def get_pivot_data(self, related_id: str) -> dict[str, Any] | None: 

69 """Return pivot data for a single attached record.""" 

70 return None 

71 

72 async def update_pivot(self, related_id: str, pivot_data: dict[str, Any]) -> None: 

73 """Update pivot data for an attached record.""" 

74 

75 async def render(self, request: Request, resource_name: str = "") -> str: 

76 items = await self.get_query() 

77 attached_ids = await self.get_attached_ids() 

78 rel_name = self.get_relationship_name() 

79 

80 rows_html = "" 

81 for item in items: 

82 item_id = str(getattr(item, "id", "")) 

83 is_attached = item_id in attached_ids 

84 label = str(getattr(item, "name", item_id)) 

85 

86 pivot_data = await self.get_pivot_data(item_id) if is_attached else None 

87 pivot_cells = self._render_pivot_cells(item_id, pivot_data) 

88 

89 checked = "checked" if is_attached else "" 

90 row = f"""<tr class="{"bg-primary-50 dark:bg-primary-900/20" if is_attached else ""}"> 

91 <td class="px-4 py-2"> 

92 <input type="checkbox" class="belongs-to-many-checkbox rounded border-border text-primary-600 focus:ring-primary-500" 

93 data-related-id="{item_id}" {checked} 

94 hx-post="/admin/{resource_name}/{self.parent_id}/relations/{rel_name}/toggle" 

95 hx-vals='{{"related_id": "{item_id}"}}' 

96 hx-target="closest tr" hx-swap="outerHTML" /> 

97 </td> 

98 <td class="px-4 py-2 text-sm text-foreground">{label}</td> 

99 {pivot_cells} 

100 <td class="px-4 py-2 text-sm text-muted-foreground">{item_id}</td> 

101 </tr>""" 

102 rows_html += row 

103 

104 header = f"""<div class="flex items-center justify-between mb-4"> 

105 <h3 class="text-lg font-medium text-foreground">{rel_name.replace("_", " ").title()}</h3> 

106 <div class="flex gap-2"> 

107 <input type="text" class="px-3 py-1.5 text-sm border rounded-lg" 

108 placeholder="Search..." id="search-{rel_name}" 

109 hx-trigger="keyup changed delay:300ms" 

110 hx-get="/admin/{resource_name}/{self.parent_id}/relations/{rel_name}" 

111 hx-target="#relation-panel-{rel_name}" hx-select=".relation-panel" /> 

112 </div> 

113 </div>""" 

114 

115 table = f"""<table class="min-w-full divide-y divide-border"> 

116 <thead class="bg-muted dark:bg-card"> 

117 <tr> 

118 <th class="px-4 py-2 text-left text-xs font-medium text-muted-foreground uppercase">Attach</th> 

119 <th class="px-4 py-2 text-left text-xs font-medium text-muted-foreground uppercase">Record</th> 

120 {self._render_pivot_headers()} 

121 <th class="px-4 py-2 text-left text-xs font-medium text-muted-foreground uppercase">ID</th> 

122 </tr> 

123 </thead> 

124 <tbody class="divide-y divide-border">{rows_html}</tbody> 

125 </table>""" 

126 

127 return f"""<div class="relation-panel p-4" id="relation-panel-{rel_name}"> 

128 {header} 

129 {table} 

130 <div class="mt-3 flex gap-2"> 

131 <button type="button" class="px-3 py-1.5 text-sm font-medium text-white bg-primary-600 rounded-lg hover:bg-primary-700" 

132 hx-post="/admin/{resource_name}/{self.parent_id}/relations/{rel_name}/sync" 

133 hx-target="#relation-panel-{rel_name}" hx-swap="outerHTML">Save</button> 

134 </div> 

135 </div>""" 

136 

137 def _render_pivot_headers(self) -> str: 

138 if not self.pivot_columns: 

139 return "" 

140 return "".join( 

141 f'<th class="px-4 py-2 text-left text-xs font-medium text-muted-foreground uppercase">{c.replace("_", " ").title()}</th>' 

142 for c in self.pivot_columns 

143 ) 

144 

145 def _render_pivot_cells( 

146 self, related_id: str, pivot_data: dict[str, Any] | None 

147 ) -> str: 

148 if not self.pivot_columns: 

149 return "" 

150 cells = "" 

151 for col in self.pivot_columns: 

152 value = (pivot_data or {}).get(col, "") 

153 cells += f"""<td class="px-4 py-2"> 

154 <input type="text" class="px-2 py-1 text-sm border rounded w-full" 

155 value="{value}" name="pivot_{col}_{related_id}" 

156 hx-post="/admin/{self.parent_id}/relations/{self.get_relationship_name()}/pivot/{related_id}" 

157 hx-trigger="change" hx-swap="none" /> 

158 </td>""" 

159 return cells 

160 

161 def get_pivot_routes(self, resource_name: str) -> list[Any]: 

162 """Return additional routes for pivot operations.""" 

163 from starlette.responses import HTMLResponse 

164 from starlette.routing import Route 

165 

166 prefix = f"/admin/{resource_name}/{self.parent_id}/relations/{self.get_relationship_name()}" 

167 

168 async def _handle_toggle(request: Any) -> Response: 

169 if request.headers.get("content-type") == "application/json": 

170 body = await request.json() 

171 else: 

172 body = request.scope.get("admin_form_data") 

173 if body is None: 

174 body = await request.form() 

175 related_id = body.get("related_id", "") 

176 attached = await self.get_attached_ids() 

177 if related_id in attached: 

178 await self.detach(related_id) 

179 else: 

180 await self.attach(related_id) 

181 return await self._render_single_row(request, resource_name, related_id) 

182 

183 async def _handle_sync(request: Any) -> Response: 

184 if request.headers.get("content-type") == "application/json": 

185 body = await request.json() 

186 else: 

187 body = request.scope.get("admin_form_data") 

188 if body is None: 

189 body = await request.form() 

190 raw_ids = body.get("related_ids", "") 

191 if isinstance(raw_ids, str): 

192 ids = ( 

193 loads_str(raw_ids) 

194 if raw_ids.startswith("[") 

195 else raw_ids.split(",") 

196 ) 

197 else: 

198 ids = raw_ids or [] 

199 await self.sync(ids) 

200 html = await self.render(request, resource_name) 

201 return HTMLResponse(html) 

202 

203 async def _handle_pivot_update(request: Any) -> Response: 

204 related_id = request.path_params.get("related_id", "") 

205 form = request.scope.get("admin_form_data") 

206 if form is None: 

207 form = await request.form() 

208 pivot_data = dict(form) 

209 await self.update_pivot(related_id, pivot_data) 

210 return HTMLResponse("") 

211 

212 return [ 

213 Route(f"{prefix}/toggle", _handle_toggle, methods=["POST"]), 

214 Route(f"{prefix}/sync", _handle_sync, methods=["POST"]), 

215 Route( 

216 f"{prefix}/pivot/{{related_id}}", _handle_pivot_update, methods=["POST"] 

217 ), 

218 ] 

219 

220 async def _render_single_row( 

221 self, request: Any, resource_name: str, related_id: str 

222 ) -> Any: 

223 from starlette.responses import HTMLResponse 

224 

225 items = await self.get_query() 

226 attached_ids = await self.get_attached_ids() 

227 item = next((i for i in items if str(getattr(i, "id", "")) == related_id), None) 

228 if item is None: 

229 return HTMLResponse("") 

230 

231 is_attached = related_id in attached_ids 

232 label = str(getattr(item, "name", related_id)) 

233 pivot_data = await self.get_pivot_data(related_id) if is_attached else None 

234 pivot_cells = self._render_pivot_cells(related_id, pivot_data) 

235 checked = "checked" if is_attached else "" 

236 

237 row = f"""<tr class="{"bg-primary-50 dark:bg-primary-900/20" if is_attached else ""}"> 

238 <td class="px-4 py-2"> 

239 <input type="checkbox" class="belongs-to-many-checkbox rounded border-border text-primary-600 focus:ring-primary-500" 

240 data-related-id="{related_id}" {checked} 

241 hx-post="/admin/{resource_name}/{self.parent_id}/relations/{self.get_relationship_name()}/toggle" 

242 hx-vals='{{"related_id": "{related_id}"}}' 

243 hx-target="closest tr" hx-swap="outerHTML" /> 

244 </td> 

245 <td class="px-4 py-2 text-sm text-foreground">{label}</td> 

246 {pivot_cells} 

247 <td class="px-4 py-2 text-sm text-muted-foreground">{related_id}</td> 

248 </tr>""" 

249 return HTMLResponse(row)