Coverage for src/lexigram/admin/data/query.py: 6%
158 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:31 +0800
1"""Unified query specification for lexigram-admin.
3This module provides the QuerySpec and PagedResult types that serve as the
4canonical query format across all admin layers (controllers, services, data sources).
6This is the single source of truth for query types. QueryBuilder and the
7old Query dataclass are deprecated in favor of QuerySpec.
8"""
10from __future__ import annotations
12from dataclasses import dataclass, field
13from enum import StrEnum
14from typing import Any, Generic, Literal, Self, TypeVar
16T = TypeVar("T")
18_UNSET: Any = object() # sentinel to distinguish "not provided" from None
21class FilterOperator(StrEnum):
22 """Supported filter operators for data queries."""
24 EQ = "eq"
25 NEQ = "neq"
26 GT = "gt"
27 GTE = "gte"
28 LT = "lt"
29 LTE = "lte"
30 IN = "in"
31 NOT_IN = "not_in"
32 CONTAINS = "contains"
33 ICONTAINS = "icontains"
34 STARTS_WITH = "starts_with"
35 ENDS_WITH = "ends_with"
36 IS_NULL = "is_null"
37 BETWEEN = "between"
40@dataclass(frozen=True)
41class FilterCondition:
42 """A single filter condition in a query."""
44 field: str
45 operator: FilterOperator
46 value: Any
49@dataclass(frozen=True)
50class QuerySpec:
51 """Unified query specification used across all layers.
53 QuerySpec provides an immutable, composable query interface that supports:
54 - Pagination (page-based and cursor-based)
55 - Sorting (single field, ascending/descending)
56 - Full-text search with configurable fields
57 - Arbitrary filters via dict
58 - Structured filter conditions with operators (where)
59 - Field selection and eager loading
60 - Grouping
62 All modification methods return a new QuerySpec instance (immutable updates).
64 Example:
65 >>> query = QuerySpec().with_page(2).with_filters(status="active")
66 >>> query = query.with_sort("created_at", "desc")
67 >>> query = query.with_search("john", fields=["name", "email"])
68 >>> query = query.with_where_eq("role", "admin")
69 """
71 # Pagination
72 page: int = 1
73 per_page: int = 20
74 cursor: str | None = None
76 # Sorting
77 sort_by: str | None = None
78 sort_order: Literal["asc", "desc"] = "asc"
80 # Search
81 search: str | None = None
82 search_fields: list[str] = field(default_factory=list)
84 # Simple filters (key=value, from URL params)
85 filters: dict[str, Any] = field(default_factory=dict)
87 # Structured filter conditions (with operators)
88 where: tuple[FilterCondition, ...] = ()
90 # Field selection
91 select_fields: tuple[str, ...] = ()
93 # Relations to eagerly load
94 include: list[str] = field(default_factory=list)
96 # Grouping
97 group_by: str | None = None
99 # Soft delete: when True, results include soft-deleted records
100 include_deleted: bool = False
102 @property
103 def offset(self) -> int:
104 """Calculate SQL offset from page and per_page."""
105 return (self.page - 1) * self.per_page
107 @property
108 def limit(self) -> int:
109 """Return per_page as limit (alias for SQL compatibility)."""
110 return self.per_page
112 @property
113 def is_cursor_based(self) -> bool:
114 """Check if this query uses cursor-based pagination."""
115 return self.cursor is not None
117 @property
118 def has_search(self) -> bool:
119 """Check if search is active."""
120 return bool(self.search and self.search_fields)
122 @property
123 def has_filters(self) -> bool:
124 """Check if any filters are active."""
125 return bool(self.filters)
127 @property
128 def has_sort(self) -> bool:
129 """Check if sorting is specified."""
130 return self.sort_by is not None
132 @property
133 def resolved_sort(self) -> tuple[str | None, Literal["asc", "desc"]]:
134 """Return (field, direction) with leading '-' prefix decoded to 'desc'."""
135 if self.sort_by and self.sort_by.startswith("-"):
136 return self.sort_by[1:], "desc"
137 return self.sort_by, self.sort_order
139 @staticmethod
140 def _condition_to_repo_key(condition: FilterCondition) -> str:
141 """Convert a FilterCondition into a repository-compatible key (e.g. ``age__gt``)."""
142 eq_types = (FilterOperator.EQ,)
143 suffix_map: dict[FilterOperator, str] = {
144 FilterOperator.NEQ: "__neq",
145 FilterOperator.GT: "__gt",
146 FilterOperator.GTE: "__gte",
147 FilterOperator.LT: "__lt",
148 FilterOperator.LTE: "__lte",
149 FilterOperator.IN: "__in",
150 FilterOperator.NOT_IN: "__not_in",
151 FilterOperator.CONTAINS: "__contains",
152 FilterOperator.ICONTAINS: "__icontains",
153 FilterOperator.STARTS_WITH: "__startswith",
154 FilterOperator.ENDS_WITH: "__endswith",
155 FilterOperator.IS_NULL: "__isnull",
156 FilterOperator.BETWEEN: "__between",
157 }
158 suffix = suffix_map.get(condition.operator)
159 if suffix is not None:
160 return f"{condition.field}{suffix}"
161 # EQ special case: list values become __in
162 if isinstance(condition.value, (list, tuple)):
163 return f"{condition.field}__in"
164 return condition.field
166 def to_repository_filters(self) -> dict[str, Any] | None:
167 """Merge ``filters`` dict and ``filter_conditions`` into repository-style dict.
169 Returns None when both are empty so callers can pass it directly as the
170 optional ``filters`` parameter without an extra None-check.
171 """
172 result: dict[str, Any] = {}
174 # Old-style dict filters
175 if self.filters:
176 result.update(self.filters)
178 # New-style filter conditions — convert operators to __suffix
179 for condition in self.where:
180 result[self._condition_to_repo_key(condition)] = condition.value
182 return result if result else None
184 @property
185 def filter_conditions(self) -> list[FilterCondition]:
186 """Combine ``where`` conditions and ``filters`` dict into a single filter list."""
187 result = list(self.where)
188 for key, value in self.filters.items():
189 result.append(
190 FilterCondition(field=key, operator=FilterOperator.EQ, value=value)
191 )
192 return result
194 def _copy(
195 self,
196 *,
197 page: int | None = None,
198 per_page: int | None = None,
199 cursor: str | None | object = _UNSET,
200 sort_by: str | None | object = _UNSET,
201 sort_order: Literal["asc", "desc"] | None = None,
202 search: str | None | object = _UNSET,
203 search_fields: list[str] | None = None,
204 filters: dict[str, Any] | None = None,
205 where: tuple[FilterCondition, ...] | None = None,
206 select_fields: tuple[str, ...] | None = None,
207 include: list[str] | None = None,
208 group_by: str | None | object = _UNSET,
209 include_deleted: bool | None = None,
210 ) -> QuerySpec:
211 """Internal helper: return a copy with selective overrides.
213 Fields whose values can legitimately be ``None`` (cursor, sort_by,
214 search, group_by) default to the sentinel ``_UNSET`` so that
215 ``None`` is treated as an explicit clear-value rather than
216 "not provided".
217 """
218 return QuerySpec(
219 page=self.page if page is None else page,
220 per_page=self.per_page if per_page is None else per_page,
221 cursor=self.cursor if cursor is _UNSET else cursor, # type: ignore[arg-type]
222 sort_by=self.sort_by if sort_by is _UNSET else sort_by, # type: ignore[arg-type]
223 sort_order=self.sort_order if sort_order is None else sort_order,
224 search=self.search if search is _UNSET else search, # type: ignore[arg-type]
225 search_fields=list(self.search_fields)
226 if search_fields is None
227 else search_fields,
228 filters=dict(self.filters) if filters is None else filters,
229 where=self.where if where is None else where,
230 select_fields=self.select_fields
231 if select_fields is None
232 else select_fields,
233 include=list(self.include) if include is None else include,
234 group_by=self.group_by if group_by is _UNSET else group_by, # type: ignore[arg-type]
235 include_deleted=self.include_deleted
236 if include_deleted is None
237 else include_deleted,
238 )
240 def with_page(self, page: int) -> QuerySpec:
241 """Return new QuerySpec with updated page.
243 Args:
244 page: Page number (1-indexed)
246 Returns:
247 New QuerySpec instance with updated page
248 """
249 return self._copy(page=max(1, page), cursor=None)
251 def with_per_page(self, per_page: int) -> QuerySpec:
252 """Return new QuerySpec with updated per_page.
254 Args:
255 per_page: Number of items per page (clamped to 1-1000)
257 Returns:
258 New QuerySpec instance with updated per_page
259 """
260 return self._copy(per_page=max(1, min(per_page, 1000)))
262 def with_cursor(self, cursor: str | None) -> QuerySpec:
263 """Return new QuerySpec with cursor-based pagination.
265 Args:
266 cursor: Cursor string for pagination
268 Returns:
269 New QuerySpec instance with cursor pagination
270 """
271 return self._copy(page=1, cursor=cursor)
273 def with_sort(
274 self,
275 field: str | None,
276 order: Literal["asc", "desc"] = "asc",
277 ) -> QuerySpec:
278 """Return new QuerySpec with updated sort.
280 Args:
281 field: Field name to sort by (None to clear sort)
282 order: Sort direction ("asc" or "desc")
284 Returns:
285 New QuerySpec instance with updated sort
286 """
287 return self._copy(sort_by=field, sort_order=order)
289 def with_search(
290 self,
291 term: str | None,
292 fields: list[str] | None = None,
293 ) -> QuerySpec:
294 """Return new QuerySpec with search parameters.
296 Args:
297 term: Search term (None to clear search)
298 fields: Fields to search in (uses existing if not provided)
300 Returns:
301 New QuerySpec instance with search
302 """
303 return self._copy(
304 page=1,
305 cursor=None,
306 search=term,
307 search_fields=fields if fields is not None else list(self.search_fields),
308 )
310 def with_filters(self, **filters: Any) -> QuerySpec:
311 """Return new QuerySpec with additional/updated filters.
313 Args:
314 **filters: Key-value filter parameters to add/update
316 Returns:
317 New QuerySpec instance with merged filters
318 """
319 merged = {**self.filters, **filters}
320 return self._copy(page=1, cursor=None, filters=merged)
322 def with_filter(self, key: str, value: Any) -> QuerySpec:
323 """Return new QuerySpec with a single filter added/updated.
325 Args:
326 key: Filter key
327 value: Filter value
329 Returns:
330 New QuerySpec instance with filter
331 """
332 return self.with_filters(**{key: value})
334 def without_filter(self, key: str) -> QuerySpec:
335 """Return new QuerySpec with a filter removed.
337 Args:
338 key: Filter key to remove
340 Returns:
341 New QuerySpec instance without the filter
342 """
343 return self._copy(
344 page=1,
345 cursor=None,
346 filters={k: v for k, v in self.filters.items() if k != key},
347 )
349 def clear_filters(self) -> QuerySpec:
350 """Return new QuerySpec with all filters cleared.
352 Returns:
353 New QuerySpec instance with no filters
354 """
355 return self._copy(page=1, cursor=None, filters={})
357 def with_deleted(self, include: bool = True) -> QuerySpec:
358 """Return new QuerySpec that includes (or excludes) soft-deleted records.
360 Args:
361 include: When True, soft-deleted records are returned alongside active ones.
363 Returns:
364 New QuerySpec instance with updated include_deleted flag.
365 """
366 return self._copy(include_deleted=include)
368 def with_include(self, *relations: str) -> QuerySpec:
369 """Return new QuerySpec with relations to eagerly load.
371 Args:
372 *relations: Relation names to include
374 Returns:
375 New QuerySpec instance with includes
376 """
377 combined = list(set(self.include) | set(relations))
378 return self._copy(include=combined)
380 # ========== Structured filter condition methods (replaces QueryBuilder) ==========
382 def with_where(
383 self,
384 field: str,
385 operator: str | FilterOperator,
386 value: Any,
387 ) -> QuerySpec:
388 """Add a structured filter condition.
390 Args:
391 field: The field name to filter on.
392 operator: The operator (e.g., "eq", FilterOperator.GT).
393 value: The value to compare against.
395 Returns:
396 New QuerySpec instance with the condition appended.
397 """
398 if isinstance(operator, str):
399 operator = FilterOperator(operator)
400 return self._copy(
401 page=1,
402 cursor=None,
403 where=(*self.where, FilterCondition(field, operator, value)),
404 )
406 def with_where_eq(self, field: str, value: Any) -> QuerySpec:
407 """Add an equality filter condition."""
408 return self.with_where(field, FilterOperator.EQ, value)
410 def with_where_in(self, field: str, values: list[Any]) -> QuerySpec:
411 """Add an IN filter condition."""
412 return self.with_where(field, FilterOperator.IN, values)
414 def with_where_contains(self, field: str, value: str) -> QuerySpec:
415 """Add a CONTAINS filter condition."""
416 return self.with_where(field, FilterOperator.CONTAINS, value)
418 def with_where_between(self, field: str, min_val: Any, max_val: Any) -> QuerySpec:
419 """Add a BETWEEN filter condition."""
420 return self.with_where(field, FilterOperator.BETWEEN, (min_val, max_val))
422 def with_order_by(self, field: str, direction: str = "asc") -> QuerySpec:
423 """Set the sort field and direction (alias for with_sort)."""
424 return self._copy(sort_by=field, sort_order=direction) # type: ignore[arg-type]
426 def with_order_by_desc(self, field: str) -> QuerySpec:
427 """Set the sort field with descending direction."""
428 return self._copy(sort_by=field, sort_order="desc")
430 def with_select(self, *fields: str) -> QuerySpec:
431 """Specify which fields to include in the result.
433 Args:
434 *fields: Field names to include.
436 Returns:
437 New QuerySpec instance with select_fields set.
438 """
439 return self._copy(select_fields=fields)
441 def with_group_by(self, field: str) -> QuerySpec:
442 """Set grouping field.
444 Args:
445 field: The field name to group by.
447 Returns:
448 New QuerySpec instance with group_by set.
449 """
450 return self._copy(group_by=field)
452 @classmethod
453 def from_dict(cls, data: dict[str, Any]) -> Self:
454 """Create QuerySpec from dictionary (e.g., query parameters).
456 Args:
457 data: Dictionary with query parameters
459 Returns:
460 New QuerySpec instance
461 """
462 return cls(
463 page=int(data.get("page", 1)),
464 per_page=int(data.get("per_page", 20)),
465 cursor=data.get("cursor"),
466 sort_by=data.get("sort_by") or data.get("sort"),
467 sort_order=data.get("sort_order", data.get("order", "asc")),
468 search=data.get("search") or data.get("q"),
469 search_fields=data.get("search_fields", []),
470 select_fields=tuple(data.get("select_fields", [])),
471 group_by=data.get("group_by"),
472 filters={
473 k: v
474 for k, v in data.items()
475 if k
476 not in (
477 "page",
478 "per_page",
479 "cursor",
480 "sort_by",
481 "sort",
482 "sort_order",
483 "order",
484 "search",
485 "q",
486 "search_fields",
487 "select_fields",
488 "group_by",
489 "include",
490 )
491 },
492 include=data.get("include", [])
493 if isinstance(data.get("include"), list)
494 else data.get("include", "").split(",")
495 if data.get("include")
496 else [],
497 include_deleted=bool(data.get("include_deleted", False)),
498 )
500 def to_dict(self) -> dict[str, Any]:
501 """Convert QuerySpec to dictionary.
503 Returns:
504 Dictionary representation (excludes None/empty values)
505 """
506 result: dict[str, Any] = {
507 "page": self.page,
508 "per_page": self.per_page,
509 }
511 if self.cursor:
512 result["cursor"] = self.cursor
513 if self.sort_by:
514 result["sort_by"] = self.sort_by
515 result["sort_order"] = self.sort_order
516 if self.search:
517 result["search"] = self.search
518 if self.search_fields:
519 result["search_fields"] = self.search_fields
520 if self.select_fields:
521 result["select_fields"] = list(self.select_fields)
522 if self.group_by:
523 result["group_by"] = self.group_by
524 if self.filters:
525 result["filters"] = self.filters
526 if self.include:
527 result["include"] = self.include
528 if self.include_deleted:
529 result["include_deleted"] = True
531 return result