Coverage for src/lexigram/admin/resources/list_renderer.py: 58%
241 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 15:04 +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 group_by=state.group_by
133 or (table_config.group_by if table_config else None),
134 empty_state_title=(
135 table_config.empty_state_title if table_config else None
136 ),
137 empty_state_message=(
138 table_config.empty_state_message if table_config else None
139 ),
140 empty_state_icon=(
141 table_config.empty_state_icon if table_config else None
142 ),
143 search_fields=getattr(resource, "search_fields", None),
144 ),
145 total=total,
146 user=user,
147 loading=False,
148 )
150 is_htmx = wants_fragment(request)
151 if is_htmx:
152 hx_target = request.headers.get("HX-Target", "")
154 # Only emit OOB control fragments for data-zone requests (search,
155 # filter, paginate) where the primary swap targets #table-data and
156 # toolbar elements outside the data zone need updating. Skip OOB
157 # for full-zone swaps (#lexigram-table) and sidebar nav
158 # (#main-content) since the primary swap replaces the entire
159 # subtree, making OOB redundant.
160 if hx_target == Zones.DATA.id:
161 dt.props["htmx_request"] = True
163 content = render_to_string(dt)
164 resp_headers = {}
166 # Synchronization: Force the browser URL to match the clean server-side state.
167 # This removes empty params (search=&foo=) that HTMX sends via hx-include.
168 # We only do this if push was not explicitly disabled in the request.
169 if request.headers.get("HX-Push-Url") != "false":
170 resp_headers["HX-Push-Url"] = state.to_url(resource_prefix)
172 return HTMLResponse(content, headers=resp_headers)
174 # Direct navigation — return full page via AdminRenderer (Jinja2 + nav population).
175 return self._renderer.render_page(
176 dt,
177 request=request,
178 title=label,
179 breadcrumbs=[
180 {"label": "Dashboard", "url": self._config.prefix},
181 {"label": label, "url": resource_prefix},
182 ],
183 )
185 async def _fetch_data(
186 self, request, resource, state: TableState, source_columns
187 ) -> Any:
188 """Fetch data from the resource service."""
189 timer = OperationTimer()
190 items: list[Any] = []
191 total = 0
192 status = "success"
193 failed = False
195 if not resource:
196 self._metrics.record_operation(
197 "list",
198 resource=self.resource_name,
199 status=status,
200 duration_seconds=timer.elapsed(),
201 )
202 return items, total
204 # Use resource search_fields when available, otherwise derive from columns
205 resource_search_fields: list[str] | None = getattr(
206 resource, "search_fields", None
207 )
208 if resource_search_fields:
209 search_fields = list(resource_search_fields)
210 else:
211 search_fields = []
212 for col in source_columns:
213 if hasattr(col, "is_searchable") and col.is_searchable():
214 search_fields.append(col.name)
216 # Extract active filters from TableState
217 filters: dict[str, Any] = dict(state.filters) if state.filters else {}
219 try:
220 # Check if this resource should be cached
221 cache_spec = self._resolve_cache_spec(resource)
223 # Check if search integration should be used
224 search_spec = self._resolve_search_spec(resource)
226 # Check if resilience integration should be used
227 resilient_spec = self._resolve_resilient_spec(resource)
229 # Search path: when a search query is active and a search spec exists,
230 # route through the search integration instead of the LIKE-based query
231 if state.search and search_spec and self._search_integration:
232 index = search_spec.index_name or self.resource_name
233 result = await self._search_integration.query(
234 index=index,
235 query_str=state.search,
236 limit=state.per_page,
237 offset=(state.page - 1) * state.per_page,
238 )
239 if hasattr(result, "rows"):
240 items = list(result.rows)
241 total = result.row_count
242 elif isinstance(result, dict):
243 items = list(result.get("results", []))
244 total = result.get("total", len(items))
245 else:
246 items, total = [], 0
248 # If search returned results, return them.
249 # Otherwise fall through to fetch_list when fallback_to_like is enabled
250 # so that the SearchQueryDataSourceWrapper can try a LIKE query.
251 if items or not self._search_integration.fallback_to_like:
252 status = "success"
253 self._metrics.record_operation(
254 "list",
255 resource=self.resource_name,
256 status=status,
257 duration_seconds=timer.elapsed(),
258 )
259 return items, total
261 logger.info(
262 "search_integration_fallback",
263 resource=self.resource_name,
264 query=state.search,
265 )
267 # Primary path: resource.fetch_list() — handles all service patterns
268 if hasattr(resource, "fetch_list"):
269 fetch_kwargs = {
270 "limit": state.per_page,
271 "offset": (state.page - 1) * state.per_page,
272 "filters": filters,
273 "search": state.search or None,
274 "search_fields": search_fields or None,
275 "sort_by": state.sort_by or None,
276 "sort_order": state.sort_order or "asc",
277 "include_deleted": state.include_deleted,
278 }
280 # Build the fetch callable, optionally wrapped with resilience
281 if resilient_spec and self._resilience_integration:
283 async def _resilient_fetch() -> tuple[list[Any], int]:
284 return await self._resilience_integration.execute(
285 resource.fetch_list,
286 **fetch_kwargs,
287 )
289 fetcher = _resilient_fetch
290 else:
292 async def _fetch() -> tuple[list[Any], int]:
293 return await resource.fetch_list(**fetch_kwargs)
295 fetcher = _fetch
297 if cache_spec and self._cache_integration:
298 cache_key = self._cache_integration.cache_key(
299 self.resource_name,
300 str(state.page),
301 state.sort_by or "",
302 )
303 items, total = await self._cache_integration.get_or_compute(
304 cache_key,
305 fetcher,
306 cache_spec.ttl_seconds,
307 )
308 else:
309 items, total = await fetcher()
311 logger.info(
312 "search.fetch_list_result",
313 resource=self.resource_name,
314 search=state.search,
315 search_fields=search_fields,
316 item_count=len(items),
317 total=total,
318 )
320 # Legacy fallback: direct service.find_many(query)
321 elif hasattr(resource, "service") and resource.service:
322 service = resource.service
323 if hasattr(service, "find_many"):
324 qs = QuerySpec()
325 if state.search and search_fields:
326 qs = qs.with_search(state.search, search_fields)
327 if state.page and state.per_page:
328 qs = qs.with_page(state.page).with_per_page(state.per_page)
329 if state.sort_by:
330 qs = qs.with_order_by(state.sort_by, state.sort_order)
331 if state.include_deleted:
332 qs = qs.with_deleted(True)
333 for field, value in filters.items():
334 if isinstance(value, list):
335 qs = qs.with_where_in(field, list(value))
336 else:
337 qs = qs.with_where_eq(field, value)
338 result = await service.find_many(qs)
339 items = result.items if hasattr(result, "items") else result
340 total = result.total if hasattr(result, "total") else len(items)
341 elif hasattr(service, "list"):
342 items = await service.list(
343 limit=state.per_page,
344 offset=(state.page - 1) * state.per_page,
345 )
346 total = len(items)
347 except DataError as e:
348 logger.error(
349 "Failed to list items for %s: %s",
350 self.resource_name,
351 e,
352 exc_info=True,
353 )
354 raise DataError(
355 message=f"Failed to retrieve {self.resource_name} items",
356 original_error=e,
357 ) from None
358 except Exception: # noqa: BLE001
359 logger.exception("admin.resource_fetch_error", resource=self.resource_name)
360 failed = True
362 if failed:
363 status = "error"
365 logger.info(
366 "search.fetch_data_result",
367 resource=self.resource_name,
368 search=state.search,
369 item_count=len(items),
370 total=total,
371 status=status,
372 failed=failed,
373 )
375 self._metrics.record_operation(
376 "list",
377 resource=self.resource_name,
378 status=status,
379 duration_seconds=timer.elapsed(),
380 )
381 return items, total
383 def _resolve_cache_spec(self, resource: Any) -> Any:
384 """Return a CacheableSpec if resource is cacheable, else None."""
385 if not resource:
386 return None
387 spec_fn = getattr(resource, "cache_spec", None)
388 if not spec_fn:
389 return None
390 spec = spec_fn()
391 if not spec:
392 return None
393 from lexigram.admin.integrations import get as get_integration
395 self._cache_integration = get_integration("CacheIntegration")
396 return spec if self._cache_integration else None
398 def _resolve_search_spec(self, resource: Any) -> Any:
399 """Return a SearchableSpec if resource is searchable, else None."""
400 if not resource:
401 return None
402 spec_fn = getattr(resource, "search_spec", None)
403 if not spec_fn:
404 return None
405 spec = spec_fn()
406 if not spec:
407 return None
408 from lexigram.admin.integrations import get as get_integration
410 self._search_integration = get_integration("SearchIntegration")
411 return spec if self._search_integration else None
413 def _resolve_resilient_spec(self, resource: Any) -> Any:
414 """Return a ResilientSpec if resource is resilient, else None."""
415 if not resource:
416 return None
417 spec_fn = getattr(resource, "resilient_spec", None)
418 if not spec_fn:
419 return None
420 spec = spec_fn()
421 if not spec:
422 return None
423 from lexigram.admin.integrations import get as get_integration
425 self._resilience_integration = get_integration("ResilienceIntegration")
426 return spec if self._resilience_integration else None
428 def _build_columns(self, source_columns, items) -> Any:
429 """Build column definitions for the data table."""
430 from lexigram.admin.ui.filters.base import Filter
432 columns = []
433 # Auto-generate columns if missing (post-fetch)
434 if not source_columns and items:
435 first_item = items[0]
436 item_dict = (
437 first_item.model_dump()
438 if hasattr(first_item, "model_dump")
439 else (
440 dict(first_item)
441 if not isinstance(first_item, (str, int, float))
442 else {"id": first_item}
443 )
444 )
445 for key in list(item_dict.keys())[:6]:
446 columns.append(
447 TextColumn(name=key, label=key.replace("_", " ").title()),
448 )
449 else:
450 # Convert or preserve columns
451 for col in source_columns:
452 if isinstance(col, Filter):
453 continue
454 if hasattr(col, "render") or isinstance(col, OrgColumn):
455 # Already a component
456 columns.append(col)
457 else:
458 # Fallback for simple objects
459 new_col = TextColumn(
460 name=getattr(col, "name", str(col)),
461 label=getattr(
462 col,
463 "label",
464 getattr(col, "name", str(col)).replace("_", " ").title(),
465 ),
466 ).sortable(True)
467 columns.append(new_col)
469 return columns
471 def _get_filter_options(self, table_config, resource) -> Any:
472 """Get filter options from resource configuration."""
473 filter_options = []
474 if table_config and table_config.filter_options:
475 filter_options = table_config.filter_options
476 elif resource and hasattr(resource, "filter_options"):
477 filter_options = (
478 resource.filter_options
479 if not callable(resource.filter_options)
480 else resource.filter_options()
481 )
482 elif resource and hasattr(resource, "filters"):
483 filter_options = (
484 resource.filters
485 if not callable(resource.filters)
486 else resource.filters()
487 )
488 return filter_options
490 def _get_row_actions(self, table_config, resource, resource_prefix) -> Any:
491 """Get row actions from resource configuration."""
492 row_actions = []
493 if table_config and table_config.actions:
494 row_actions = table_config.actions
495 elif resource and hasattr(resource, "actions"):
496 actions = (
497 resource.actions
498 if not callable(resource.actions)
499 else resource.actions()
500 )
501 row_actions = list(actions)
503 # Inject default URLs for standard actions if missing
504 from lexigram.ui.actions.standard import EditAction, ViewAction
506 for action in row_actions:
507 if (
508 isinstance(action, (EditAction, ViewAction))
509 and not action.get_url()
510 and not action.get_hx_get()
511 ):
512 # Default logic: {prefix}/{id}/edit or {prefix}/{id}
513 postfix = "/edit" if isinstance(action, EditAction) else ""
515 # Use hx_get for partial updates (SlideOver/Modal)
516 action.hx(get=f"{resource_prefix.rstrip('/')}/{{id}}{postfix}")
518 return row_actions
520 def _get_header_actions(self, table_config, resource) -> Any:
521 """Get header actions from resource configuration."""
522 header_actions = []
523 if (
524 table_config
525 and hasattr(table_config, "header_actions")
526 and table_config.header_actions
527 ):
528 header_actions = table_config.header_actions
529 elif resource and hasattr(resource, "header_actions"):
530 header_actions = (
531 resource.header_actions
532 if not callable(resource.header_actions)
533 else resource.header_actions()
534 )
535 return header_actions
537 def _get_bulk_actions(self, table_config, resource) -> Any:
538 """Get bulk actions from resource configuration."""
539 bulk_actions_list = []
540 source_bulk = []
541 if table_config and table_config.bulk_actions:
542 source_bulk = table_config.bulk_actions
543 elif resource and hasattr(resource, "bulk_actions"):
544 source_bulk = (
545 resource.bulk_actions
546 if not callable(resource.bulk_actions)
547 else resource.bulk_actions()
548 )
550 from lexigram.ui.actions.standard import BulkAction as OrgBulkAction
551 from lexigram.ui.actions.standard import DeleteBulkAction
553 for ba in source_bulk:
554 if isinstance(ba, OrgBulkAction):
555 bulk_actions_list.append(ba)
556 elif ba == "delete_selected":
557 bulk_actions_list.append(DeleteBulkAction(label="Delete Selected"))
558 elif isinstance(ba, str):
559 # Generic bulk action from string
560 bulk_actions_list.append(
561 OrgBulkAction(label=ba.replace("_", " ").title(), name=ba),
562 )
564 return bulk_actions_list
567__all__ = ["ListRenderer"]