Coverage for src / lexigram / admin / services / filter_manager.py: 0%
210 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-11 02:25 +0800
1"""Advanced filter management for DataTable with presets, URL sharing, and validation."""
3from __future__ import annotations
5from collections.abc import Awaitable, Callable
6from dataclasses import dataclass
7import inspect
8from typing import Any
9from urllib.parse import parse_qs, urlencode
11from lexigram.logging import get_logger
12from lexigram.serialization import dumps_str, loads_str
14logger = get_logger(__name__)
17@dataclass
18class FilterPreset:
19 """Saved filter configuration."""
21 name: str
22 filters: dict[str, Any]
23 user_id: int | None = None
24 is_default: bool = False
25 is_shared: bool = False
26 created_at: str | None = None
29@dataclass
30class FilterDefinition:
31 """Definition of a filter field using a concrete filter instance."""
33 field: str
34 filter: Any | None = None
35 label: str | None = None
36 default_value: Any = None
37 validator: Callable[[Any], bool | Awaitable[bool]] | None = None
38 required: bool = False
41class FilterManager:
42 """Manage DataTable filters with presets, URL encoding, validation, and search translation."""
44 def __init__(
45 self,
46 filter_definitions: list[FilterDefinition] | None = None,
47 translator: Any | None = None,
48 ) -> None:
49 """Initialize filter manager.
51 Args:
52 filter_definitions: List of available filter definitions.
53 translator: ``FilterSetTranslator`` used to convert the active
54 filter dict into a :class:`~lexigram.search.engine.SearchQuery`.
55 Defaults to a bare :class:`FilterSetTranslator` instance when
56 ``None`` and ``lexigram-search`` is installed (it is stateless,
57 so the default is safe). Pass ``None`` when search is not needed.
58 """
59 self.filter_definitions = {f.field: f for f in (filter_definitions or [])}
60 self._presets: dict[str, FilterPreset] = {}
61 if translator is not None:
62 self._translator: Any = translator
63 else:
64 self._translator = None
66 async def process_request(
67 self,
68 query_params: Any,
69 state_filters: dict[str, Any] | None = None,
70 ) -> tuple[dict[str, Any], set[str]]:
71 """
72 Process request query parameters into a validated filter dictionary (async).
73 """
74 filters: dict[str, Any] = {}
75 consumed: set[str] = set()
77 for field, definition in self.filter_definitions.items():
78 # Support both explicit Filter types and raw definitions
79 f_inst = definition.filter
80 raw_val = None
82 if f_inst and hasattr(f_inst, "get_value_from_request"):
83 try:
84 # Check if it's an async method
85 if inspect.iscoroutinefunction(f_inst.get_value_from_request):
86 raw_val = await f_inst.get_value_from_request(query_params)
87 else:
88 raw_val = f_inst.get_value_from_request(query_params)
89 except (
90 ConnectionError,
91 RuntimeError,
92 ValueError,
93 TypeError,
94 AttributeError,
95 ):
96 raw_val = None
98 # Fallback to direct query param access if not handled by filter instance
99 if raw_val is None:
100 raw_val = query_params.get(field)
102 if raw_val is not None and raw_val != "":
103 # Handle default exclusion
104 default = None
105 if hasattr(f_inst, "get_default"):
106 default = f_inst.get_default() # type: ignore[union-attr]
107 elif definition.default_value is not None:
108 default = definition.default_value
110 if raw_val != default:
111 # Parse typed value
112 try:
113 if hasattr(f_inst, "from_url_param"):
114 parsed = f_inst.from_url_param(raw_val) # type: ignore[union-attr]
115 else:
116 # Try JSON or raw
117 try:
118 parsed = loads_str(raw_val)
119 except (ValueError, TypeError):
120 parsed = raw_val
121 filters[field] = parsed
122 except (ValueError, TypeError):
123 filters[field] = raw_val
125 # Track consumed params
126 if hasattr(f_inst, "get_consumed_params"):
127 consumed.update(f_inst.get_consumed_params()) # type: ignore[union-attr]
128 else:
129 consumed.add(field)
131 # Merge anonymous filters from state if provided
132 if state_filters:
133 filters |= {k: v for k, v in state_filters.items() if k not in consumed}
135 return filters, consumed
137 async def validate_filters(
138 self,
139 filters: dict[str, Any],
140 ) -> tuple[bool, dict[str, str]]:
141 """
142 Validate filter values against definitions (async).
143 """
144 errors: dict[str, str] = {}
146 for field, value in filters.items():
147 if field not in self.filter_definitions:
148 errors[field] = f"Unknown filter field: {field}"
149 continue
151 definition = self.filter_definitions[field]
153 # Check required
154 if definition.required and (value is None or value == ""):
155 errors[field] = f"{definition.label or field} is required"
156 continue
158 # Skip validation if empty and not required
159 if value is None or value == "":
160 continue
162 # Options validation via concrete Filter instances
163 if definition.filter is not None and hasattr(
164 definition.filter,
165 "get_options",
166 ):
167 try:
168 if inspect.iscoroutinefunction(definition.filter.get_options):
169 opts = await definition.filter.get_options()
170 else:
171 opts = definition.filter.get_options()
172 except Exception: # noqa: BLE001 — user-supplied callable; must catch broadly
173 logger.exception(
174 "Failed to get options for filter %s; skipping options validation",
175 field,
176 )
177 opts = []
179 # Support list or dict option storage (only validate if opts available)
180 if opts:
181 if isinstance(value, list):
182 invalid = list(filter(lambda v: v not in opts, value))
183 if invalid:
184 errors[field] = (
185 f"Invalid value for {definition.label or field}"
186 )
187 continue
188 elif value not in opts:
189 errors[field] = f"Invalid value for {definition.label or field}"
190 continue
192 # Custom validator (catch exceptions from validator)
193 if definition.validator:
194 try:
195 if inspect.iscoroutinefunction(definition.validator):
196 valid = await definition.validator(value)
197 else:
198 valid = definition.validator(value)
199 except Exception: # noqa: BLE001 — user-supplied callable; must catch broadly
200 logger.exception("Validator raised exception for filter %s", field)
201 valid = False
203 if not valid:
204 errors[field] = f"Invalid value for {definition.label or field}"
205 continue
207 return len(errors) == 0, errors
209 def encode_to_url(self, filters: dict[str, Any]) -> str:
210 """
211 Encode filters to URL query string.
212 """
213 # Remove empty values
214 clean_filters = {k: v for k, v in filters.items() if v is not None and v != ""}
216 # Convert to query string
217 params = {}
218 for key, value in clean_filters.items():
219 if isinstance(value, (list, dict)):
220 params[f"filter[{key}]"] = dumps_str(value)
221 else:
222 params[f"filter[{key}]"] = str(value)
224 return urlencode(params)
226 def decode_from_url(self, query_string: str) -> dict[str, Any]:
227 """
228 Decode filters from URL query string.
229 """
230 params = parse_qs(query_string.lstrip("?"))
231 filters: dict[str, Any] = {}
233 for key, values in params.items():
234 if key.startswith("filter[") and key.endswith("]"):
235 field = key[7:-1] # Extract field name
236 value = values[0] if values else None
238 # Try to parse JSON for complex values
239 if value:
240 try:
241 filters[field] = loads_str(value)
242 except (ValueError, TypeError):
243 filters[field] = value
245 return filters
247 def get_default_filters(self) -> dict[str, Any]:
248 """
249 Get default filter values from definitions.
250 """
251 defaults = {}
252 for field, definition in self.filter_definitions.items():
253 if definition.default_value is not None:
254 defaults[field] = definition.default_value
255 return defaults
257 async def create_preset(
258 self,
259 name: str,
260 filters: dict[str, Any],
261 user_id: int | None = None,
262 is_default: bool = False,
263 is_shared: bool = False,
264 ) -> FilterPreset:
265 """
266 Create a filter preset (async).
267 """
268 from datetime import datetime
270 preset = FilterPreset(
271 name=name,
272 filters=filters.copy(),
273 user_id=user_id,
274 is_default=is_default,
275 is_shared=is_shared,
276 created_at=datetime.now().isoformat(),
277 )
279 # Generate preset key
280 key = f"{user_id}:{name}" if user_id else name
281 self._presets[key] = preset
283 return preset
285 async def get_preset(
286 self,
287 name: str,
288 user_id: int | None = None,
289 ) -> FilterPreset | None:
290 """
291 Get a saved preset (async).
292 """
293 key = f"{user_id}:{name}" if user_id else name
294 return self._presets.get(key)
296 async def list_presets(
297 self,
298 user_id: int | None = None,
299 include_shared: bool = True,
300 ) -> list[FilterPreset]:
301 """
302 List available presets (async).
303 """
304 presets = []
306 for preset in self._presets.values():
307 # User-specific presets
308 if (
309 (user_id and preset.user_id == user_id)
310 or (include_shared and preset.is_shared)
311 or preset.user_id is None
312 ):
313 presets.append(preset)
315 return presets
317 async def delete_preset(self, name: str, user_id: int | None = None) -> bool:
318 """
319 Delete a preset (async).
320 """
321 key = f"{user_id}:{name}" if user_id else name
322 if key in self._presets:
323 del self._presets[key]
324 return True
325 return False
327 def get_active_filter_badges(self, filters: dict[str, Any]) -> list[dict[str, Any]]:
328 """
329 Get active filter information for display as badges.
330 """
331 badges = []
333 for field, value in filters.items():
334 if value is None or value == "":
335 continue
337 definition = self.filter_definitions.get(field)
338 label = definition.label if definition else field
340 # Format value for display
341 if isinstance(value, list):
342 display_value = ", ".join(str(v) for v in value)
343 elif isinstance(value, dict):
344 display_value = dumps_str(value)
345 else:
346 display_value = str(value)
348 # Determine badge type from filter class name if possible
349 if definition and getattr(definition, "filter", None):
350 badge_type = definition.filter.__class__.__name__.lower()
351 else:
352 badge_type = "text"
354 badges.append(
355 {
356 "field": field,
357 "label": label,
358 "value": display_value,
359 "type": badge_type,
360 },
361 )
363 return badges
365 def clear_all_filters(self) -> dict[str, Any]:
366 """
367 Clear all filters and return defaults.
368 """
369 return self.get_default_filters()
371 def has_active_filters(self, filters: dict[str, Any]) -> bool:
372 """
373 Check if any filters are currently active.
374 """
375 defaults = self.get_default_filters()
377 for field, value in filters.items():
378 if value is None or value == "":
379 continue
381 # Check against default
382 if field in defaults:
383 if value != defaults[field]:
384 return True
385 else:
386 return True
388 return False
390 async def apply_preset(
391 self,
392 name: str,
393 user_id: int | None = None,
394 ) -> dict[str, Any] | None:
395 """Apply a saved preset and return filter values (async)."""
396 preset = await self.get_preset(name, user_id)
397 if preset:
398 return preset.filters.copy()
399 return None
401 # ------------------------------------------------------------------
402 # Search integration
403 # ------------------------------------------------------------------
405 def to_filter_set(
406 self,
407 filters: dict[str, Any],
408 *,
409 search_query: str | None = None,
410 order_by: str | None = None,
411 order_dir: str = "asc",
412 page: int = 1,
413 page_size: int = 25,
414 ) -> dict[str, Any]:
415 """Convert an active filter dict to a contract-neutral filter-set payload.
417 Each entry in *filters* is mapped to a condition payload using
418 operator inference:
420 * ``list`` value → ``FilterOperator.IN``
421 * ``None`` value → ``FilterOperator.IS_NULL``
422 * all other values → ``FilterOperator.EQ``
424 Empty strings and ``None`` values are dropped (same convention as
425 :meth:`encode_to_url`).
427 Args:
428 filters: Active filter dict (field → value) as produced by
429 :meth:`process_request`.
430 search_query: Optional free-text query forwarded to
431 :attr:`~lexigram.search.filterset.FilterSet.search_query`.
432 order_by: Field name to sort by.
433 order_dir: Sort direction — ``"asc"`` (default) or ``"desc"``.
434 page: 1-based page number.
435 page_size: Results per page.
437 Returns:
438 A dictionary payload with conditions and paging/sorting metadata,
439 ready for ``to_search_query`` translation.
440 """
441 conditions: list[dict[str, Any]] = []
442 for field, value in filters.items():
443 if value is None:
444 conditions.append(
445 {"field": field, "operator": "is_null", "value": None}
446 )
447 elif value == "":
448 continue
449 elif isinstance(value, list):
450 conditions.append({"field": field, "operator": "in", "value": value})
451 else:
452 conditions.append({"field": field, "operator": "eq", "value": value})
453 return {
454 "conditions": tuple(conditions),
455 "order_by": order_by,
456 "order_dir": order_dir,
457 "page": page,
458 "page_size": page_size,
459 "search_query": search_query,
460 }
462 def to_search_query(
463 self,
464 filters: dict[str, Any],
465 *,
466 search_query: str | None = None,
467 order_by: str | None = None,
468 order_dir: str = "asc",
469 page: int = 1,
470 page_size: int = 25,
471 ) -> dict[str, Any]:
472 """Translate the active filter dict directly into a search-query payload.
474 Convenience wrapper that calls :meth:`to_filter_set` followed by
475 ``translator.translate()`` when a translator was provided.
477 Args:
478 filters: Active filter dict (field → value).
479 search_query: Optional free-text query string.
480 order_by: Field name to sort by.
481 order_dir: Sort direction — ``"asc"`` or ``"desc"``.
482 page: 1-based page number.
483 page_size: Results per page.
485 Returns:
486 A search-query payload ready to pass to a search backend.
488 Raises:
489 RuntimeError: If no translator was configured.
490 """
491 if self._translator is None:
492 raise RuntimeError(
493 "FilterManager.to_search_query() requires a translator object "
494 "injected via the constructor.",
495 )
496 filter_set = self.to_filter_set(
497 filters,
498 search_query=search_query,
499 order_by=order_by,
500 order_dir=order_dir,
501 page=page,
502 page_size=page_size,
503 )
504 return self._translator.translate(filter_set)
507__all__ = [
508 "FilterDefinition",
509 "FilterManager",
510 "FilterPreset",
511]