Coverage for src/lexigram/admin/dashboard/page_filters.py: 96%
85 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Page-level dashboard filter state (session-persisted) and form rendering.
3Implements the Filament ``HasFiltersForm``/``InteractsWithPageFilters`` parity
4gap for the admin dashboard: a page declares a filter schema
5(``PageFilterField`` list), and this module owns reading/merging that state
6(schema defaults → session → query params, query wins), persisting it per page
7in the request session, rendering the apply/reset filter form, and building
8widget fetch URLs annotated with the current filter values.
9"""
11from __future__ import annotations
13from typing import Any
14from urllib.parse import urlencode
16from lexigram.contracts.admin.types import PageFilterField
17from lexigram.ui import el
19_SESSION_PREFIX = "admin_page_filters."
20_RESET_PARAM = "reset_page_filters"
23def _session(request: Any) -> Any:
24 """Return the request session, or ``None`` when no session support exists."""
25 return getattr(request, "session", None)
28def _coerce(field: PageFilterField, raw: str) -> Any:
29 """Coerce a query-param string to the field's declared type."""
30 if field.type == "boolean":
31 return raw in ("1", "true", "on")
32 if field.type == "number":
33 try:
34 return int(raw)
35 except ValueError:
36 return raw
37 return raw
40def read_page_filters(
41 request: Any,
42 page_name: str,
43 schema: list[PageFilterField] | tuple[PageFilterField, ...],
44) -> dict[str, Any]:
45 """Merge persisted state with request inputs for a page's filter schema.
47 Resolution order (later wins): schema defaults, then session state, then
48 query params. A ``reset_page_filters`` query param clears the session state
49 and returns only schema defaults.
51 Args:
52 request: Asgi request exposing ``session`` and ``query_params``.
53 page_name: Stable per-page key for session persistence.
54 schema: Declared filter fields for the page.
56 Returns:
57 Mapping of ``{field_name: value}`` for every schema field.
58 """
59 merged: dict[str, Any] = {}
60 for field in schema:
61 if field.default is not None:
62 merged[field.name] = field.default
64 key = f"{_SESSION_PREFIX}{page_name}"
65 session = _session(request)
66 if session is not None:
67 stored = session.get(key)
68 if isinstance(stored, dict):
69 merged.update({k: v for k, v in stored.items() if k in merged})
71 params = request.query_params
72 if params.get(_RESET_PARAM) in ("1", "true"):
73 clear_page_filters(request, page_name)
74 return {f.name: f.default for f in schema if f.default is not None}
76 for field in schema:
77 if field.name in params:
78 merged[field.name] = _coerce(field, params[field.name])
79 return merged
82def save_page_filters(request: Any, page_name: str, values: dict[str, Any]) -> None:
83 """Persist filter values for a page in the request session.
85 Args:
86 request: Asgi request exposing ``session``.
87 page_name: Per-page persistence key.
88 values: Filter values to store.
89 """
90 session = _session(request)
91 if session is not None:
92 session[f"{_SESSION_PREFIX}{page_name}"] = dict(values)
95def clear_page_filters(request: Any, page_name: str) -> None:
96 """Drop persisted filter values for a page from the session.
98 Args:
99 request: Asgi request exposing ``session``.
100 page_name: Per-page persistence key.
101 """
102 session = _session(request)
103 if session is not None:
104 session.pop(f"{_SESSION_PREFIX}{page_name}", None)
107def applied_from_query(
108 request: Any, schema: list[PageFilterField] | tuple[PageFilterField, ...]
109) -> bool:
110 """Return whether any query param targeted a declared filter field.
112 Args:
113 request: Asgi request exposing ``query_params``.
114 schema: Declared filter fields for the page.
116 Returns:
117 ``True`` if at least one schema field name appears in the query params.
118 """
119 return any(f.name in request.query_params for f in schema)
122def widget_fetch_url(
123 endpoint: str,
124 page_filters: dict[str, Any] | None,
125) -> str:
126 """Append current page filter values to a widget fetch URL.
128 Args:
129 endpoint: The widget's ``hx-get`` render endpoint.
130 page_filters: Current filter values, or ``None``.
132 Returns:
133 The endpoint with filter values appended as query parameters.
134 """
135 if not page_filters:
136 return endpoint
137 pairs = {k: v for k, v in page_filters.items() if v is not None and v != ""}
138 if not pairs:
139 return endpoint
140 encoded = urlencode(pairs)
141 sep = "&" if "?" in endpoint else "?"
142 return f"{endpoint}{sep}{encoded}"
145def render_page_filter_form(
146 schema: list[PageFilterField] | tuple[PageFilterField, ...],
147 current: dict[str, Any],
148 action_url: str,
149) -> Any:
150 """Render an apply/reset filter form for a page.
152 Args:
153 schema: Declared filter fields for the page.
154 current: Current filter values (used to prefill the controls).
155 action_url: GET target for apply (the page URL itself).
157 Returns:
158 An ``el`` form node, or ``None`` when there is nothing to render.
159 """
160 if not schema:
161 return None
163 fields: list[Any] = []
164 for field in schema:
165 value = current.get(field.name, field.default)
166 common_attrs: dict[str, Any] = {
167 "name": field.name,
168 "class": "bg-card border border-border rounded-md px-2 py-1 text-sm text-foreground focus:outline-none focus:ring-1 focus:ring-primary",
169 }
170 if field.description:
171 common_attrs["title"] = field.description
172 if field.type == "select" and field.options:
173 options = []
174 for option_value, option_label in field.options:
175 option_attrs: dict[str, Any] = {}
176 if value is not None and str(option_value) == str(value):
177 option_attrs["selected"] = True
178 options.append(
179 el(
180 "option",
181 option_label,
182 value=str(option_value),
183 **option_attrs,
184 )
185 )
186 input_el = el("select", *options, **common_attrs)
187 elif field.type == "boolean":
188 input_el = el(
189 "input",
190 type="checkbox",
191 checked=bool(value),
192 **common_attrs,
193 )
194 elif field.type == "number":
195 input_el = el(
196 "input",
197 type="number",
198 value=str(value) if value is not None else "",
199 **common_attrs,
200 )
201 else:
202 input_el = el(
203 "input",
204 type="text",
205 value=str(value) if value is not None else "",
206 **common_attrs,
207 )
208 fields.append(
209 el(
210 "label",
211 el(
212 "span",
213 field.label,
214 class_="block text-xs text-muted-foreground mb-1",
215 ),
216 input_el,
217 class_="block",
218 )
219 )
221 fields.extend(
222 [
223 el(
224 "button",
225 "Apply",
226 type="submit",
227 class_="px-3 py-1 text-sm rounded-md bg-primary text-primary-foreground hover:opacity-90",
228 ),
229 el(
230 "a",
231 "Reset",
232 href=_with_reset_param(action_url),
233 class_="px-3 py-1 text-sm rounded-md border border-border text-muted-foreground hover:bg-muted",
234 ),
235 ]
236 )
238 return el(
239 "form",
240 *fields,
241 method="get",
242 action=action_url,
243 class_="chart-filters flex flex-wrap gap-3 items-end",
244 )
247def _with_reset_param(action_url: str) -> str:
248 """Return ``action_url`` carrying the reset filter query param."""
249 sep = "&" if "?" in action_url else "?"
250 return f"{action_url}{sep}{_RESET_PARAM}=1"
253__all__ = [
254 "applied_from_query",
255 "clear_page_filters",
256 "read_page_filters",
257 "render_page_filter_form",
258 "save_page_filters",
259 "widget_fetch_url",
260]