Coverage for src / lexigram / admin / resources / list_renderer.py: 0%
238 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:07 +0800
1from __future__ import annotations
3"""List view rendering for admin resources."""
5from typing import Any
7from starlette.responses import HTMLResponse
9from lexigram.admin.config import AdminConfig
10from lexigram.admin.data.query import QuerySpec
11from lexigram.admin.engine.renderer import AdminRenderer
12from lexigram.admin.exceptions import DataError
13from lexigram.admin.observability.admin_metrics import AdminMetrics, OperationTimer
14from lexigram.admin.resources.config import TableConfiguration
15from lexigram.admin.state.context import wants_fragment
16from lexigram.admin.ui.organisms.data_table import DataTable
17from lexigram.di.decorators import inject
18from lexigram.logging import get_logger
19from lexigram.ui import Column as OrgColumn
20from lexigram.ui import TableState, TextColumn, Zones, render_to_string
22logger = get_logger(__name__)
25@inject
26class ListRenderer:
27 """Handles rendering of list views for admin resources."""
29 def __init__(
30 self,
31 config: AdminConfig,
32 resource_name: str,
33 renderer: AdminRenderer,
34 metrics: AdminMetrics | None = None,
35 ):
36 self._config = config
37 self.resource_name = resource_name
38 self._renderer = renderer
39 self._metrics = metrics or AdminMetrics(None)
40 self._cache_integration: Any = None
41 self._search_integration: Any = None
42 self._resilience_integration: Any = None
44 async def render(
45 self,
46 request,
47 resource,
48 user=None,
49 ) -> HTMLResponse:
50 """Render list view with DataTable component."""
51 # Get resource configuration
52 table_config = (
53 resource.get_table_config()
54 if resource and hasattr(resource, "get_table_config")
55 else None
56 )
57 label = (
58 (table_config.resource_name if table_config else self.resource_name)
59 .replace("_", " ")
60 .title()
61 )
62 resource_prefix = f"{self._config.prefix}/{self.resource_name}"
64 # Resolve Columns Early for Search
65 source_columns = []
66 if table_config and table_config.columns:
67 source_columns = table_config.columns
68 elif resource and hasattr(resource, "columns"):
69 # Check if columns is a property/method
70 source_columns = (
71 resource.columns
72 if not callable(resource.columns)
73 else resource.columns()
74 )
76 # Parse request params using TableState
77 state = TableState.from_request(
78 request,
79 defaults={
80 "sort_by": table_config.default_sort_by if table_config else None,
81 "sort_order": table_config.default_sort_order
82 if table_config
83 else "asc",
84 "view": table_config.default_view if table_config else "tabular",
85 "layout": table_config.default_layout if table_config else "stack",
86 "per_page": table_config.per_page if table_config else 20,
87 }
88 if table_config
89 else {},
90 )
92 # Map list_view sort params to TableState params
93 if request.query_params.get("sort"):
94 state.sort_by = request.query_params.get("sort")
95 if request.query_params.get("dir"):
96 state.sort_order = request.query_params.get("dir")
98 # Fetch data from service
99 items, total = await self._fetch_data(request, resource, state, source_columns)
101 # Build columns
102 columns = self._build_columns(source_columns, items)
104 # Prepare Filters
105 filter_options = self._get_filter_options(table_config, resource)
107 # Prepare Actions
108 row_actions = self._get_row_actions(table_config, resource, resource_prefix)
110 header_actions = self._get_header_actions(table_config, resource)
112 # Prepare Bulk Actions
113 bulk_actions_list = self._get_bulk_actions(table_config, resource)
115 # Prepare DataTable
116 dt = DataTable(
117 columns=columns,
118 data=items,
119 state=state,
120 config=TableConfiguration(
121 columns=columns,
122 resource_name=self.resource_name,
123 resource_prefix=resource_prefix,
124 actions=row_actions,
125 header_actions=header_actions,
126 bulk_actions=bulk_actions_list,
127 filter_options=filter_options,
128 default_sort_by=state.sort_by,
129 default_sort_order=state.sort_order,
130 default_layout=table_config.default_layout if table_config else "stack",
131 default_view=table_config.default_view if table_config else "tabular",
132 search_fields=getattr(resource, "search_fields", None),
133 ),
134 total=total,
135 user=user,
136 loading=False,
137 )
139 is_htmx = wants_fragment(request)
140 if is_htmx:
141 hx_target = request.headers.get("HX-Target", "")
143 # Only emit OOB control fragments for data-zone requests (search,
144 # filter, paginate) where the primary swap targets #table-data and
145 # toolbar elements outside the data zone need updating. Skip OOB
146 # for full-zone swaps (#lexigram-table) and sidebar nav
147 # (#main-content) since the primary swap replaces the entire
148 # subtree, making OOB redundant.
149 if hx_target == Zones.DATA.id:
150 dt.props["htmx_request"] = True
152 content = render_to_string(dt)
153 resp_headers = {}
155 # Synchronization: Force the browser URL to match the clean server-side state.
156 # This removes empty params (search=&foo=) that HTMX sends via hx-include.
157 # We only do this if push was not explicitly disabled in the request.
158 if request.headers.get("HX-Push-Url") != "false":
159 resp_headers["HX-Push-Url"] = state.to_url(resource_prefix)
161 return HTMLResponse(content, headers=resp_headers)
163 # Direct navigation — return full page via AdminRenderer (Jinja2 + nav population).
164 return self._renderer.render_page(
165 dt,
166 request=request,
167 title=label,
168 breadcrumbs=[
169 {"label": "Dashboard", "url": self._config.prefix},
170 {"label": label, "url": resource_prefix},
171 ],
172 )
174 async def _fetch_data(
175 self, request, resource, state: TableState, source_columns
176 ) -> Any:
177 """Fetch data from the resource service."""
178 timer = OperationTimer()
179 items: list[Any] = []
180 total = 0
181 status = "success"
182 failed = False
184 if not resource:
185 self._metrics.record_operation(
186 "list",
187 resource=self.resource_name,
188 status=status,
189 duration_seconds=timer.elapsed(),
190 )
191 return items, total
193 # Use resource search_fields when available, otherwise derive from columns
194 resource_search_fields: list[str] | None = getattr(
195 resource, "search_fields", None
196 )
197 if resource_search_fields:
198 search_fields = list(resource_search_fields)
199 else:
200 search_fields = []
201 for col in source_columns:
202 if hasattr(col, "is_searchable") and col.is_searchable():
203 search_fields.append(col.name)
205 # Extract active filters from TableState
206 filters: dict[str, Any] = dict(state.filters) if state.filters else {}
208 try:
209 # Check if this resource should be cached
210 cache_spec = self._resolve_cache_spec(resource)
212 # Check if search integration should be used
213 search_spec = self._resolve_search_spec(resource)
215 # Check if resilience integration should be used
216 resilient_spec = self._resolve_resilient_spec(resource)
218 # Search path: when a search query is active and a search spec exists,
219 # route through the search integration instead of the LIKE-based query
220 if state.search and search_spec and self._search_integration:
221 index = search_spec.index_name or self.resource_name
222 result = await self._search_integration.query(
223 index=index,
224 query_str=state.search,
225 limit=state.per_page,
226 offset=(state.page - 1) * state.per_page,
227 )
228 if hasattr(result, "rows"):
229 items = list(result.rows)
230 total = result.row_count
231 elif isinstance(result, dict):
232 items = list(result.get("results", []))
233 total = result.get("total", len(items))
234 else:
235 items, total = [], 0
237 # If search returned results, return them.
238 # Otherwise fall through to fetch_list when fallback_to_like is enabled
239 # so that the SearchQueryDataSourceWrapper can try a LIKE query.
240 if items or not self._search_integration.fallback_to_like:
241 status = "success"
242 self._metrics.record_operation(
243 "list",
244 resource=self.resource_name,
245 status=status,
246 duration_seconds=timer.elapsed(),
247 )
248 return items, total
250 logger.info(
251 "search_integration_fallback",
252 resource=self.resource_name,
253 query=state.search,
254 )
256 # Primary path: resource.fetch_list() — handles all service patterns
257 if hasattr(resource, "fetch_list"):
258 fetch_kwargs = {
259 "limit": state.per_page,
260 "offset": (state.page - 1) * state.per_page,
261 "filters": filters,
262 "search": state.search or None,
263 "search_fields": search_fields or None,
264 "sort_by": state.sort_by or None,
265 "sort_order": state.sort_order or "asc",
266 "include_deleted": state.include_deleted,
267 }
269 # Build the fetch callable, optionally wrapped with resilience
270 if resilient_spec and self._resilience_integration:
272 async def _resilient_fetch() -> tuple[list[Any], int]:
273 return await self._resilience_integration.execute(
274 resource.fetch_list,
275 **fetch_kwargs,
276 )
278 fetcher = _resilient_fetch
279 else:
281 async def _fetch() -> tuple[list[Any], int]:
282 return await resource.fetch_list(**fetch_kwargs)
284 fetcher = _fetch
286 if cache_spec and self._cache_integration:
287 cache_key = self._cache_integration.cache_key(
288 self.resource_name,
289 str(state.page),
290 state.sort_by or "",
291 )
292 items, total = await self._cache_integration.get_or_compute(
293 cache_key,
294 fetcher,
295 cache_spec.ttl_seconds,
296 )
297 else:
298 items, total = await fetcher()
300 logger.info(
301 "search.fetch_list_result",
302 resource=self.resource_name,
303 search=state.search,
304 search_fields=search_fields,
305 item_count=len(items),
306 total=total,
307 )
309 # Legacy fallback: direct service.find_many(query)
310 elif hasattr(resource, "service") and resource.service:
311 service = resource.service
312 if hasattr(service, "find_many"):
313 qs = QuerySpec()
314 if state.search and search_fields:
315 qs = qs.with_search(state.search, search_fields)
316 if state.page and state.per_page:
317 qs = qs.with_page(state.page).with_per_page(state.per_page)
318 if state.sort_by:
319 qs = qs.with_order_by(state.sort_by, state.sort_order)
320 if state.include_deleted:
321 qs = qs.with_deleted(True)
322 for field, value in filters.items():
323 if isinstance(value, list):
324 qs = qs.with_where_in(field, list(value))
325 else:
326 qs = qs.with_where_eq(field, value)
327 result = await service.find_many(qs)
328 items = result.items if hasattr(result, "items") else result
329 total = result.total if hasattr(result, "total") else len(items)
330 elif hasattr(service, "list"):
331 items = await service.list(
332 limit=state.per_page,
333 offset=(state.page - 1) * state.per_page,
334 )
335 total = len(items)
336 except DataError as e:
337 logger.error(
338 "Failed to list items for %s: %s",
339 self.resource_name,
340 e,
341 exc_info=True,
342 )
343 raise DataError(
344 message=f"Failed to retrieve {self.resource_name} items",
345 original_error=e,
346 ) from None
347 except Exception: # noqa: BLE001
348 logger.exception("admin.resource_fetch_error", resource=self.resource_name)
349 failed = True
351 if failed:
352 status = "error"
354 logger.info(
355 "search.fetch_data_result",
356 resource=self.resource_name,
357 search=state.search,
358 item_count=len(items),
359 total=total,
360 status=status,
361 failed=failed,
362 )
364 self._metrics.record_operation(
365 "list",
366 resource=self.resource_name,
367 status=status,
368 duration_seconds=timer.elapsed(),
369 )
370 return items, total
372 def _resolve_cache_spec(self, resource: Any) -> Any:
373 """Return a CacheableSpec if resource is cacheable, else None."""
374 if not resource:
375 return None
376 spec_fn = getattr(resource, "cache_spec", None)
377 if not spec_fn:
378 return None
379 spec = spec_fn()
380 if not spec:
381 return None
382 from lexigram.admin.integrations import get as get_integration
384 self._cache_integration = get_integration("CacheIntegration")
385 return spec if self._cache_integration else None
387 def _resolve_search_spec(self, resource: Any) -> Any:
388 """Return a SearchableSpec if resource is searchable, else None."""
389 if not resource:
390 return None
391 spec_fn = getattr(resource, "search_spec", None)
392 if not spec_fn:
393 return None
394 spec = spec_fn()
395 if not spec:
396 return None
397 from lexigram.admin.integrations import get as get_integration
399 self._search_integration = get_integration("SearchIntegration")
400 return spec if self._search_integration else None
402 def _resolve_resilient_spec(self, resource: Any) -> Any:
403 """Return a ResilientSpec if resource is resilient, else None."""
404 if not resource:
405 return None
406 spec_fn = getattr(resource, "resilient_spec", None)
407 if not spec_fn:
408 return None
409 spec = spec_fn()
410 if not spec:
411 return None
412 from lexigram.admin.integrations import get as get_integration
414 self._resilience_integration = get_integration("ResilienceIntegration")
415 return spec if self._resilience_integration else None
417 def _build_columns(self, source_columns, items) -> Any:
418 """Build column definitions for the data table."""
419 columns = []
420 # Auto-generate columns if missing (post-fetch)
421 if not source_columns and items:
422 first_item = items[0]
423 item_dict = (
424 first_item.model_dump()
425 if hasattr(first_item, "model_dump")
426 else (
427 dict(first_item)
428 if not isinstance(first_item, (str, int, float))
429 else {"id": first_item}
430 )
431 )
432 for key in list(item_dict.keys())[:6]:
433 columns.append(
434 TextColumn(name=key, label=key.replace("_", " ").title()),
435 )
436 else:
437 # Convert or preserve columns
438 for col in source_columns:
439 if hasattr(col, "render") or isinstance(col, OrgColumn):
440 # Already a component
441 columns.append(col)
442 else:
443 # Fallback for simple objects
444 new_col = TextColumn(
445 name=getattr(col, "name", str(col)),
446 label=getattr(
447 col,
448 "label",
449 getattr(col, "name", str(col)).replace("_", " ").title(),
450 ),
451 ).sortable(True)
452 columns.append(new_col)
454 return columns
456 def _get_filter_options(self, table_config, resource) -> Any:
457 """Get filter options from resource configuration."""
458 filter_options = []
459 if table_config and table_config.filter_options:
460 filter_options = table_config.filter_options
461 elif resource and hasattr(resource, "filter_options"):
462 filter_options = (
463 resource.filter_options
464 if not callable(resource.filter_options)
465 else resource.filter_options()
466 )
467 elif resource and hasattr(resource, "filters"):
468 filter_options = (
469 resource.filters
470 if not callable(resource.filters)
471 else resource.filters()
472 )
473 return filter_options
475 def _get_row_actions(self, table_config, resource, resource_prefix) -> Any:
476 """Get row actions from resource configuration."""
477 row_actions = []
478 if table_config and table_config.actions:
479 row_actions = table_config.actions
480 elif resource and hasattr(resource, "actions"):
481 actions = (
482 resource.actions
483 if not callable(resource.actions)
484 else resource.actions()
485 )
486 row_actions = list(actions)
488 # Inject default URLs for standard actions if missing
489 from lexigram.ui.actions.standard import EditAction, ViewAction
491 for action in row_actions:
492 if (
493 isinstance(action, (EditAction, ViewAction))
494 and not action.get_url()
495 and not action.get_hx_get()
496 ):
497 # Default logic: {prefix}/{id}/edit or {prefix}/{id}
498 postfix = "/edit" if isinstance(action, EditAction) else ""
500 # Use hx_get for partial updates (SlideOver/Modal)
501 action.hx(get=f"{resource_prefix.rstrip('/')}/{{id}}{postfix}")
503 return row_actions
505 def _get_header_actions(self, table_config, resource) -> Any:
506 """Get header actions from resource configuration."""
507 header_actions = []
508 if (
509 table_config
510 and hasattr(table_config, "header_actions")
511 and table_config.header_actions
512 ):
513 header_actions = table_config.header_actions
514 elif resource and hasattr(resource, "header_actions"):
515 header_actions = (
516 resource.header_actions
517 if not callable(resource.header_actions)
518 else resource.header_actions()
519 )
520 return header_actions
522 def _get_bulk_actions(self, table_config, resource) -> Any:
523 """Get bulk actions from resource configuration."""
524 bulk_actions_list = []
525 source_bulk = []
526 if table_config and table_config.bulk_actions:
527 source_bulk = table_config.bulk_actions
528 elif resource and hasattr(resource, "bulk_actions"):
529 source_bulk = (
530 resource.bulk_actions
531 if not callable(resource.bulk_actions)
532 else resource.bulk_actions()
533 )
535 from lexigram.ui.actions.standard import BulkAction as OrgBulkAction
536 from lexigram.ui.actions.standard import DeleteBulkAction
538 for ba in source_bulk:
539 if isinstance(ba, OrgBulkAction):
540 bulk_actions_list.append(ba)
541 elif ba == "delete_selected":
542 bulk_actions_list.append(DeleteBulkAction(label="Delete Selected"))
543 elif isinstance(ba, str):
544 # Generic bulk action from string
545 bulk_actions_list.append(
546 OrgBulkAction(label=ba.replace("_", " ").title(), name=ba),
547 )
549 return bulk_actions_list
552__all__ = ["ListRenderer"]