Coverage for src / lexigram / admin / ui / state.py: 19%
238 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-11 02:25 +0800
1from __future__ import annotations
3import contextlib
4from dataclasses import dataclass
5from typing import Any, ClassVar, Literal
7from htpy import input as htpy_input
9from lexigram.domain import DomainModel
10from lexigram.logging import get_logger
11from lexigram.validation import Field
13logger = get_logger(__name__)
16@dataclass(init=False)
17class TableState(DomainModel):
18 """
19 Encapsulates the complete state of a DataTable.
20 This state is derived from the URL and drives the UI rendering.
21 """
23 search: str = ""
24 sort_by: str | None = None
25 sort_order: Literal["asc", "desc"] = "asc"
26 page: int = 1
27 per_page: int = 20
28 cursor: str | None = None
29 filters: dict[str, Any] = Field(default_factory=dict)
30 view: Literal["tabular", "grid", "calendar", "stacked"] = "tabular"
31 layout: Literal["sidebar", "stack"] = "stack"
32 column_order: list[str] | None = None
33 group_by: str | None = None
34 collapsed_groups: list[str] = Field(default_factory=list)
35 include_deleted: bool = False
37 # Internal defaults for clean URL generation (not part of model fields)
38 _defaults: ClassVar[dict] = {}
40 def __init__(self, **data: Any) -> None:
41 super().__init__(**data)
42 if "_defaults" not in self.__dict__:
43 object.__setattr__(self, "_defaults", {})
45 @classmethod
46 def from_request(cls, request, defaults: dict | None = None) -> TableState:
47 """
48 Create state from a request object (Starlette/ASGI compatible).
49 """
50 defaults = defaults or {}
52 # Use request.query_params directly for better multi-value handling.
53 # .get() on Starlette QueryParams returns the LAST value by default,
54 # which is what we want for overrides (e.g. link ?view=grid overriding hidden input view=tabular).
55 q = request.query_params
57 # Extract Standard Fields
58 search = q.get("search") or ""
59 sort_by = q.get("sort_by") or defaults.get("sort_by")
60 sort_order = q.get("sort_order") or defaults.get("sort_order", "asc")
62 try:
63 page = int(q.get("page", 1))
64 page = max(page, 1)
65 except (ValueError, TypeError):
66 page = 1
68 try:
69 per_page = int(q.get("per_page", 20))
70 except (ValueError, TypeError):
71 per_page = 20
72 # Explicitly check for 'limit' as an alias often used in APIs
73 with contextlib.suppress(ValueError, TypeError):
74 per_page = int(q.get("limit", 20))
76 view = q.get("data_view") or defaults.get("view", "tabular")
77 layout = q.get("layout_type") or defaults.get("layout", "stack")
78 cursor = q.get("cursor") or None
80 include_deleted_raw = q.get("include_deleted", "false")
81 include_deleted = include_deleted_raw.lower() == "true"
83 # Defensive: resolve defaults (guard against inconsistent test defaults)
84 default_sort_by = (
85 defaults.get("sort_by")
86 if isinstance(defaults.get("sort_by"), str)
87 else None
88 )
89 default_sort_order = (
90 defaults.get("sort_order", "asc")
91 if isinstance(defaults.get("sort_order", "asc"), str)
92 else "asc"
93 )
94 default_view = (
95 defaults.get("view", "tabular")
96 if isinstance(defaults.get("view", "tabular"), str)
97 else "tabular"
98 )
99 # Normalize defaults with defensive checks and logging (guard against misconfigured app defaults)
100 raw_default_layout = defaults.get("layout", "stack")
101 if isinstance(raw_default_layout, str):
102 if raw_default_layout in ("sidebar", "stack"):
103 default_layout = raw_default_layout
104 else:
105 logger.warning(
106 "Unknown default layout '%s' provided; using fallback 'stack'",
107 raw_default_layout,
108 )
109 default_layout = "stack"
110 else:
111 logger.warning(
112 "Non-string default layout provided (%r); using fallback 'stack'",
113 raw_default_layout,
114 )
115 default_layout = "stack"
117 # Apply defaults/coercion
118 if not isinstance(sort_by, (str, type(None))):
119 logger.warning("Invalid default sort_by %r; ignoring", sort_by)
120 sort_by = default_sort_by
122 if sort_order not in ("asc", "desc"):
123 logger.warning(
124 "Invalid default sort_order %r; using '%s'",
125 sort_order,
126 default_sort_order,
127 )
128 sort_order = default_sort_order
130 # Normalize view default
131 raw_default_view = defaults.get("view", "tabular")
132 if isinstance(raw_default_view, str) and raw_default_view in (
133 "tabular",
134 "grid",
135 "calendar",
136 "stacked",
137 ):
138 default_view = raw_default_view
139 else:
140 if not isinstance(raw_default_view, str):
141 logger.warning(
142 "Non-string default view provided (%r); using 'tabular'",
143 raw_default_view,
144 )
145 else:
146 logger.warning(
147 "Unknown default view '%s' provided; using 'tabular'",
148 raw_default_view,
149 )
150 default_view = "tabular"
152 # Normalize incoming 'view' param
153 if not isinstance(view, str) or view not in (
154 "tabular",
155 "grid",
156 "calendar",
157 "stacked",
158 ):
159 logger.debug(
160 "Invalid or missing request view %r; using default %r",
161 view,
162 default_view,
163 )
164 view = default_view
166 # GuardProtocol the final layout value
167 if not isinstance(layout, str) or layout not in (
168 "sidebar",
169 "stack",
170 ):
171 logger.debug(
172 "Invalid or missing request layout %r; using default %r",
173 layout,
174 default_layout,
175 )
176 layout = default_layout
178 # Extract Filters
179 # We assume any param not in this blocklist is a filter
180 known_keys = {
181 "collapsed_groups",
182 "col_order",
183 "data_view",
184 "filters",
185 "flash_message",
186 "flash_type",
187 "group_by",
188 "hx-current-url",
189 "hx-request",
190 "hx-target",
191 "hx-trigger",
192 "ids",
193 "include_deleted",
194 "layout_type",
195 "limit",
196 "next",
197 "page",
198 "per_page",
199 "render_fragment",
200 "search",
201 "select_all",
202 "sort_by",
203 "sort_order",
204 }
206 col_order_raw = q.get("col_order")
207 col_order = (
208 col_order_raw.split(",") if col_order_raw else defaults.get("column_order")
209 )
211 group_by = q.get("group_by", defaults.get("group_by"))
212 collapsed_raw = q.get("collapsed_groups", "")
213 collapsed_groups = collapsed_raw.split(",") if collapsed_raw else []
215 # For filters, we also want the last value if duplicated
216 def _coerce_value(val: str) -> Any:
217 # Normalize booleans
218 if isinstance(val, str):
219 low = val.lower()
220 if low == "true":
221 return True
222 if low == "false":
223 return False
224 # Try integer
225 try:
226 if val.isdigit() or (val.startswith("-") and val[1:].isdigit()):
227 return int(val)
228 except (ValueError, TypeError):
229 pass
230 # Try float
231 try:
232 if "." in val:
233 return float(val)
234 except (ValueError, TypeError):
235 pass
236 return val
238 filters = {}
239 for k in q:
240 if k in known_keys:
241 continue
243 filter_key = k[7:] if k.startswith("filter_") else k
245 # Support both Starlette QueryParams (with getlist) and plain dicts
246 if hasattr(q, "getlist"):
247 values = q.getlist(k)
248 else:
249 v = q.get(k)
250 values = [v] if v is not None else []
252 if not values:
253 continue
255 # Filter out empty strings
256 values = list(filter(lambda v: v is not None and v != "", values))
257 if not values:
258 continue
260 # Deduplicate values
261 unique_values = []
262 seen = set()
263 for v in values:
264 # If it's a string representation of a list, repair it
265 if isinstance(v, str) and v.startswith("[") and v.endswith("]"):
266 import ast
268 try:
269 parsed = ast.literal_eval(v)
270 if isinstance(parsed, list):
271 for item in parsed:
272 val = _coerce_value(str(item))
273 if val not in seen:
274 unique_values.append(val)
275 seen.add(val)
276 continue
277 except (TypeError, ValueError):
278 pass
280 val = _coerce_value(v)
281 if val not in seen:
282 unique_values.append(val)
283 seen.add(val)
284 values = unique_values
286 if len(values) > 1:
287 filters[filter_key] = values
288 elif values:
289 filters[filter_key] = values[0]
291 state = cls(
292 search=search,
293 sort_by=sort_by,
294 sort_order=sort_order,
295 page=page,
296 per_page=per_page,
297 cursor=cursor,
298 filters=filters,
299 view=view,
300 layout=layout,
301 column_order=col_order,
302 group_by=group_by,
303 collapsed_groups=collapsed_groups,
304 include_deleted=include_deleted,
305 )
306 object.__setattr__(state, "_defaults", defaults or {})
307 return state
309 def to_query_params(self, exclude: list[str] | None = None) -> dict:
310 """
311 Export state to dictionary suitable for URL generation.
312 Only includes non-default and non-empty values to keep URLs clean.
313 """
314 exclude = exclude or []
315 params = {}
317 # Mapping of query param names to state attribute names for default lookup
318 key_map = {
319 "data_view": "view",
320 "layout_type": "layout",
321 }
323 def add(key, val, default=None) -> Any:
324 if key in exclude:
325 return
327 # Use resource defaults if available for cleaner URLs
328 # Map query key to internal state key for defaults lookup
329 state_key = key_map.get(key, key)
330 res_default = self._defaults.get(state_key, default)
332 # Stringify for comparison if needed
333 val_str = str(val) if val is not None else ""
334 def_str = str(res_default) if res_default is not None else ""
336 # Only add if strictly non-default and non-empty
337 if val is not None and val_str != def_str and val != "":
338 params[key] = val
340 add("search", self.search, "")
341 add("page", self.page, 1)
342 add("per_page", self.per_page, 20)
343 add("sort_by", self.sort_by, "")
344 add("sort_order", self.sort_order, "asc")
345 add("data_view", self.view, "tabular")
346 add("layout_type", self.layout, "stack")
348 if self.cursor:
349 add("cursor", self.cursor)
350 if self.column_order:
351 add("col_order", ",".join(self.column_order))
352 if self.group_by:
353 add("group_by", self.group_by)
354 if self.collapsed_groups:
355 add("collapsed_groups", ",".join(self.collapsed_groups))
356 add("include_deleted", self.include_deleted, False)
358 # Add filters with prefix (non-empty only)
359 for k, v in self.filters.items():
360 param_key = f"filter_{k}"
361 if (
362 param_key not in exclude
363 and k not in exclude
364 and v is not None
365 and v != ""
366 ):
367 params[param_key] = v
369 return params
371 def to_url(self, base_path: str = "") -> str:
372 """Return a canonical URL (path + query) for this TableState."""
373 from urllib.parse import urlencode
375 params = self.to_query_params()
376 if not base_path:
377 base_path = ""
378 query = urlencode(params, doseq=True)
379 url = base_path or ""
380 if query:
381 url = f"{url}?{query}"
382 return url
384 def model_copy(self, *args, **kwargs) -> TableState:
385 """Override model_copy to preserve internal defaults."""
386 new_state = super().model_copy(*args, **kwargs)
387 object.__setattr__(new_state, "_defaults", getattr(self, "_defaults", {}))
388 return new_state
390 # These methods return NEW TableState instances with modified values.
391 # The original state is not modified (immutable pattern).
393 def with_page(self, page: int) -> TableState:
394 """
395 Return a copy with a new page number.
397 Resets cursor for offset-based pagination.
399 Example:
400 new_state = state.with_page(2)
401 attrs = HTMXAttrs.for_data_refresh(new_state, prefix)
402 """
403 return self.model_copy(update={"page": page, "cursor": None})
405 def with_per_page(self, per_page: int) -> TableState:
406 """
407 Return a copy with a new per_page value.
409 Resets to page 1 since row counts change.
410 """
411 return self.model_copy(update={"per_page": per_page, "page": 1, "cursor": None})
413 def with_search(self, search: str) -> TableState:
414 """
415 Return a copy with a new search term.
417 Resets to page 1 since results change.
418 """
419 return self.model_copy(update={"search": search, "page": 1, "cursor": None})
421 def with_filter(self, key: str, value: Any) -> TableState:
422 """
423 Return a copy with an updated filter value.
425 Resets to page 1 since results change.
427 Example:
428 new_state = state.with_filter("status", "active")
429 """
430 new_filters = {**self.filters, key: value}
431 return self.model_copy(
432 update={"filters": new_filters, "page": 1, "cursor": None},
433 )
435 def without_filter(self, key: str) -> TableState:
436 """
437 Return a copy with a filter removed.
439 Resets to page 1 since results change.
440 """
441 new_filters = {k: v for k, v in self.filters.items() if k != key}
442 return self.model_copy(
443 update={"filters": new_filters, "page": 1, "cursor": None},
444 )
446 def with_sort(self, column: str) -> TableState:
447 """
448 Return a copy with sort toggled on the given column.
450 If already sorting by this column, toggles direction.
451 Otherwise, sets ascending sort on the column.
453 Example:
454 new_state = state.with_sort("name") # asc
455 new_state = new_state.with_sort("name") # desc
456 """
457 if self.sort_by == column:
458 new_order = "desc" if self.sort_order == "asc" else "asc"
459 return self.model_copy(update={"sort_order": new_order})
460 return self.model_copy(update={"sort_by": column, "sort_order": "asc"})
462 def with_view(
463 self,
464 view: Literal["tabular", "grid", "calendar", "stacked"],
465 ) -> TableState:
466 """Return a copy with a new view type."""
467 return self.model_copy(update={"view": view})
469 def with_layout(self, layout: Literal["sidebar", "stack"]) -> TableState:
470 """Return a copy with a new layout type."""
471 return self.model_copy(update={"layout": layout})
473 def with_include_deleted(self, include_deleted: bool) -> TableState:
474 """Return a copy with a new include_deleted value.
476 Resets to page 1 since results change.
478 Example:
479 new_state = state.with_include_deleted(True)
480 """
481 return self.model_copy(
482 update={"include_deleted": include_deleted, "page": 1, "cursor": None},
483 )
485 def clear_filters(self) -> TableState:
486 """
487 Return a copy with all filters and search cleared.
489 Resets to page 1.
490 """
491 return self.model_copy(
492 update={
493 "filters": {},
494 "search": "",
495 "page": 1,
496 "cursor": None,
497 },
498 )
500 def clear_sort(self) -> TableState:
501 """Return a copy with sorting cleared."""
502 return self.model_copy(update={"sort_by": None, "sort_order": "asc"})
504 def set_resource_prefix(self, prefix: str) -> None:
505 """Set the resource prefix for URL generation."""
506 object.__setattr__(self, "_resource_prefix", prefix)
508 def get_resource_prefix(self) -> str | None:
509 """Get the resource prefix for URL generation."""
510 return getattr(self, "_resource_prefix", None)
512 # === Hidden Input Rendering ===
514 def render_hidden_inputs(self, exclude: list[str] | None = None) -> list:
515 """
516 Render hidden inputs for state preservation.
518 Used as a fallback when baked URLs aren't possible (e.g., form submissions
519 that need to preserve table state).
521 These inputs should be placed INSIDE the TABLE zone.
523 Args:
524 exclude: Optional list of field names to skip (avoid duplication with UI inputs)
526 Returns:
527 List of htpy input elements
528 """
529 inputs = []
530 excluded_keys = set(exclude or [])
532 params = self.to_query_params()
533 for key, value in params.items():
534 if key in excluded_keys:
535 continue
537 # Handle multi-value fields by rendering multiple hidden inputs
538 values = value if isinstance(value, (list, tuple)) else [value]
539 for v in values:
540 inputs.append(
541 htpy_input(
542 type="hidden",
543 name=key,
544 value=str(v),
545 data_state="true", # Mark as state input for debugging
546 ),
547 )
549 return inputs