Coverage for src / dataknobs_data / query.py: 29%
314 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-26 15:45 -0700
1"""Query construction and filtering for database operations.
3This module provides classes for building queries with filters, operators,
4sorting, pagination, and vector similarity search for database operations.
5"""
7from __future__ import annotations
9from dataclasses import dataclass, field
10from enum import Enum
11from typing import TYPE_CHECKING, Any
13if TYPE_CHECKING:
14 from collections.abc import Callable
16 import numpy as np
18 from .query_logic import ComplexQuery
19 from .vector.types import DistanceMetric
22class Operator(Enum):
23 """Query operators for filtering.
25 Operators used to build filter conditions in queries. Supports comparison,
26 pattern matching, existence checks, and range queries.
28 Example:
29 ```python
30 from dataknobs_data import Filter, Operator, Query
32 # Equality
33 filter_eq = Filter("age", Operator.EQ, 30)
35 # Comparison
36 filter_gt = Filter("score", Operator.GT, 90)
38 # Pattern matching (SQL LIKE)
39 filter_like = Filter("name", Operator.LIKE, "A%") # Names starting with 'A'
41 # IN operator
42 filter_in = Filter("status", Operator.IN, ["active", "pending"])
44 # Range query
45 filter_between = Filter("age", Operator.BETWEEN, [20, 40])
47 # Build query
48 query = Query(filters=[filter_gt, filter_like])
49 ```
50 """
52 EQ = "=" # Equal
53 NEQ = "!=" # Not equal
54 GT = ">" # Greater than
55 GTE = ">=" # Greater than or equal
56 LT = "<" # Less than
57 LTE = "<=" # Less than or equal
58 IN = "in" # In list
59 NOT_IN = "not_in" # Not in list
60 LIKE = "like" # String pattern matching (SQL LIKE)
61 NOT_LIKE = "not_like" # String pattern not matching (SQL NOT LIKE)
62 REGEX = "regex" # Regular expression matching
63 EXISTS = "exists" # Field exists
64 NOT_EXISTS = "not_exists" # Field does not exist
65 BETWEEN = "between" # Value between two bounds (inclusive)
66 NOT_BETWEEN = "not_between" # Value not between two bounds
69class SortOrder(Enum):
70 """Sort order for query results."""
72 ASC = "asc"
73 DESC = "desc"
76@dataclass
77class Filter:
78 """Represents a filter condition.
80 A Filter combines a field name, an operator, and a value to create a query condition.
81 Multiple filters can be combined in a Query for complex filtering.
83 Attributes:
84 field: The field name to filter on
85 operator: The comparison operator
86 value: The value to compare against (optional for EXISTS/NOT_EXISTS operators)
88 Example:
89 ```python
90 from dataknobs_data import Filter, Operator, Query, database_factory
92 # Create filters
93 age_filter = Filter("age", Operator.GT, 25)
94 name_filter = Filter("name", Operator.LIKE, "A%")
95 status_filter = Filter("status", Operator.IN, ["active", "pending"])
97 # Use in query
98 query = Query(filters=[age_filter, name_filter])
100 # Search database
101 db = database_factory("memory")
102 results = db.search(query)
103 ```
104 """
106 field: str
107 operator: Operator
108 value: Any = None
110 def matches(self, record_value: Any) -> bool:
111 """Check if a record value matches this filter.
113 Supports type-aware comparisons for ranges and special handling
114 for datetime/date objects.
115 """
116 if self.operator == Operator.EXISTS:
117 return record_value is not None
118 elif self.operator == Operator.NOT_EXISTS:
119 return record_value is None
120 elif record_value is None:
121 return False
123 if self.operator == Operator.EQ:
124 return record_value == self.value
125 elif self.operator == Operator.NEQ:
126 return record_value != self.value
127 elif self.operator == Operator.GT:
128 return self._compare_values(record_value, self.value, lambda a, b: a > b)
129 elif self.operator == Operator.GTE:
130 return self._compare_values(record_value, self.value, lambda a, b: a >= b)
131 elif self.operator == Operator.LT:
132 return self._compare_values(record_value, self.value, lambda a, b: a < b)
133 elif self.operator == Operator.LTE:
134 return self._compare_values(record_value, self.value, lambda a, b: a <= b)
135 elif self.operator == Operator.IN:
136 return record_value in self.value
137 elif self.operator == Operator.NOT_IN:
138 return record_value not in self.value
139 elif self.operator == Operator.BETWEEN:
140 if not isinstance(self.value, (list, tuple)) or len(self.value) != 2:
141 return False
142 lower, upper = self.value
143 return self._compare_values(record_value, lower, lambda a, b: a >= b) and \
144 self._compare_values(record_value, upper, lambda a, b: a <= b)
145 elif self.operator == Operator.NOT_BETWEEN:
146 if not isinstance(self.value, (list, tuple)) or len(self.value) != 2:
147 return True
148 lower, upper = self.value
149 return not (self._compare_values(record_value, lower, lambda a, b: a >= b) and \
150 self._compare_values(record_value, upper, lambda a, b: a <= b))
151 elif self.operator == Operator.LIKE:
152 if not isinstance(record_value, str):
153 return False
154 import re
156 pattern = self.value.replace("%", ".*").replace("_", ".")
157 return bool(re.match(f"^{pattern}$", record_value))
158 elif self.operator == Operator.NOT_LIKE:
159 if not isinstance(record_value, str):
160 return False
161 import re
163 pattern = self.value.replace("%", ".*").replace("_", ".")
164 return not bool(re.match(f"^{pattern}$", record_value))
165 elif self.operator == Operator.REGEX:
166 if not isinstance(record_value, str):
167 return False
168 import re
170 return bool(re.search(self.value, record_value))
171 else:
172 # This should never be reached as all operators are handled above
173 raise ValueError(f"Unknown operator: {self.operator}")
175 def _compare_values(self, a: Any, b: Any, comparator) -> bool:
176 """Compare two values with type awareness.
178 Handles special cases:
179 - Datetime strings are parsed for comparison
180 - Mixed numeric types are converted appropriately
181 - String comparisons are case-sensitive
182 """
183 from datetime import date, datetime
185 # Handle datetime/date comparisons
186 if isinstance(a, str) and isinstance(b, (datetime, date)):
187 try:
188 a = datetime.fromisoformat(a.replace("Z", "+00:00"))
189 except (ValueError, AttributeError):
190 return False
191 elif isinstance(b, str) and isinstance(a, (datetime, date)):
192 try:
193 b = datetime.fromisoformat(b.replace("Z", "+00:00"))
194 except (ValueError, AttributeError):
195 return False
196 elif isinstance(a, str) and isinstance(b, str):
197 # Check if both look like dates
198 if "T" in a or "-" in a:
199 try:
200 a = datetime.fromisoformat(a.replace("Z", "+00:00"))
201 b = datetime.fromisoformat(b.replace("Z", "+00:00"))
202 except (ValueError, AttributeError):
203 pass # Keep as strings
205 # Handle numeric comparisons
206 if isinstance(a, (int, float)) and isinstance(b, (int, float)):
207 return comparator(a, b)
209 # Try direct comparison
210 try:
211 return comparator(a, b)
212 except TypeError:
213 # Types not comparable
214 return False
216 def to_dict(self) -> dict[str, Any]:
217 """Convert filter to dictionary representation."""
218 return {"field": self.field, "operator": self.operator.value, "value": self.value}
220 @classmethod
221 def from_dict(cls, data: dict[str, Any]) -> Filter:
222 """Create filter from dictionary representation."""
223 return cls(
224 field=data["field"], operator=Operator(data["operator"]), value=data.get("value")
225 )
228@dataclass
229class SortSpec:
230 """Represents a sort specification."""
232 field: str
233 order: SortOrder = SortOrder.ASC
235 def to_dict(self) -> dict[str, str]:
236 """Convert sort spec to dictionary representation."""
237 return {"field": self.field, "order": self.order.value}
239 @classmethod
240 def from_dict(cls, data: dict[str, str]) -> SortSpec:
241 """Create sort spec from dictionary representation."""
242 return cls(field=data["field"], order=SortOrder(data.get("order", "asc")))
245@dataclass
246class VectorQuery:
247 """Represents a vector similarity search query.
249 This dataclass encapsulates all parameters needed for vector similarity search,
250 including the query vector, distance metric, and various search options.
251 """
253 vector: np.ndarray | list[float] # Query vector or embeddings
254 field_name: str = "embedding" # Vector field name to search
255 k: int = 10 # Number of results (top-k)
256 metric: DistanceMetric | str = "cosine" # Distance metric
257 include_source: bool = True # Include source text in results
258 score_threshold: float | None = None # Minimum similarity score
259 rerank: bool = False # Whether to rerank results
260 rerank_k: int | None = None # Number of results to rerank (default: 2*k)
261 metadata: dict[str, Any] = field(default_factory=dict) # Additional metadata
263 def to_dict(self) -> dict[str, Any]:
264 """Convert vector query to dictionary representation."""
265 import numpy as np
267 # Handle vector serialization
268 vector_data = self.vector
269 if isinstance(vector_data, np.ndarray):
270 vector_data = vector_data.tolist()
272 # Handle metric serialization
273 metric_value = self.metric
274 if hasattr(metric_value, 'value'): # DistanceMetric enum
275 metric_value = metric_value.value
277 result = {
278 "vector": vector_data,
279 "field": self.field_name,
280 "k": self.k,
281 "metric": metric_value,
282 "include_source": self.include_source,
283 }
285 if self.score_threshold is not None:
286 result["score_threshold"] = self.score_threshold
287 if self.rerank:
288 result["rerank"] = self.rerank
289 if self.rerank_k is not None:
290 result["rerank_k"] = self.rerank_k
291 if self.metadata:
292 result["metadata"] = self.metadata
294 return result
296 @classmethod
297 def from_dict(cls, data: dict[str, Any]) -> VectorQuery:
298 """Create vector query from dictionary representation."""
299 import numpy as np
301 from .vector.types import DistanceMetric
303 # Handle vector deserialization
304 vector_data = data["vector"]
305 if not isinstance(vector_data, np.ndarray):
306 vector_data = np.array(vector_data, dtype=np.float32)
308 # Handle metric deserialization
309 metric_value = data.get("metric", "cosine")
310 if isinstance(metric_value, str):
311 try:
312 metric_value = DistanceMetric(metric_value)
313 except ValueError:
314 # Keep as string if not a valid enum value
315 pass
317 return cls(
318 vector=vector_data,
319 field_name=data.get("field", "embedding"),
320 k=data.get("k", 10),
321 metric=metric_value,
322 include_source=data.get("include_source", True),
323 score_threshold=data.get("score_threshold"),
324 rerank=data.get("rerank", False),
325 rerank_k=data.get("rerank_k"),
326 metadata=data.get("metadata", {}),
327 )
330@dataclass
331class Query:
332 """Represents a database query with filters, sorting, pagination, and vector search.
334 A Query combines multiple filter conditions, sort specifications, and pagination
335 options to retrieve records from a database. Supports fluent interface for building queries.
337 Attributes:
338 filters: List of filter conditions
339 sort_specs: List of sort specifications
340 limit_value: Maximum number of results
341 offset_value: Number of results to skip
342 fields: List of field names to include (projection)
343 vector_query: Optional vector similarity search parameters
345 Example:
346 ```python
347 from dataknobs_data import Query, Filter, Operator, SortOrder, SortSpec, database_factory
349 # Simple query with filters
350 query = Query(
351 filters=[
352 Filter("age", Operator.GT, 25),
353 Filter("status", Operator.EQ, "active")
354 ]
355 )
357 # Using fluent interface
358 query = (
359 Query()
360 .filter("age", Operator.GT, 25)
361 .filter("status", Operator.EQ, "active")
362 .sort_by("age", SortOrder.DESC)
363 .limit(10)
364 .offset(20)
365 )
367 # With field projection
368 query = (
369 Query()
370 .filter("age", Operator.GT, 25)
371 .select("name", "age", "email")
372 )
374 # Execute query
375 db = database_factory("memory")
376 results = db.search(query)
377 ```
378 """
380 filters: list[Filter] = field(default_factory=list)
381 sort_specs: list[SortSpec] = field(default_factory=list)
382 limit_value: int | None = None
383 offset_value: int | None = None
384 fields: list[str] | None = None # Field projection
385 vector_query: VectorQuery | None = None # Vector similarity search
387 @property
388 def sort_property(self) -> list[SortSpec]:
389 """Get sort specifications (backward compatibility)."""
390 return self.sort_specs
392 @property
393 def limit_property(self) -> int | None:
394 """Get limit value (backward compatibility)."""
395 return self.limit_value
397 @property
398 def offset_property(self) -> int | None:
399 """Get offset value (backward compatibility)."""
400 return self.offset_value
402 def filter(self, field: str, operator: str | Operator, value: Any = None) -> Query:
403 """Add a filter to the query (fluent interface).
405 Args:
406 field: The field name to filter on
407 operator: The operator (string or Operator enum)
408 value: The value to compare against
410 Returns:
411 Self for method chaining
412 """
413 if isinstance(operator, str):
414 op_map = {
415 "=": Operator.EQ,
416 "==": Operator.EQ,
417 "!=": Operator.NEQ,
418 ">": Operator.GT,
419 ">=": Operator.GTE,
420 "<": Operator.LT,
421 "<=": Operator.LTE,
422 "in": Operator.IN,
423 "IN": Operator.IN,
424 "not_in": Operator.NOT_IN,
425 "NOT IN": Operator.NOT_IN,
426 "like": Operator.LIKE,
427 "LIKE": Operator.LIKE,
428 "regex": Operator.REGEX,
429 "exists": Operator.EXISTS,
430 "not_exists": Operator.NOT_EXISTS,
431 "between": Operator.BETWEEN,
432 "BETWEEN": Operator.BETWEEN,
433 "not_between": Operator.NOT_BETWEEN,
434 "NOT BETWEEN": Operator.NOT_BETWEEN,
435 }
436 operator = op_map.get(operator, Operator.EQ)
438 self.filters.append(Filter(field=field, operator=operator, value=value))
439 return self
441 def sort_by(self, field: str, order: str | SortOrder = "asc") -> Query:
442 """Add a sort specification to the query (fluent interface).
444 Args:
445 field: The field name to sort by
446 order: The sort order ("asc", "desc", or SortOrder enum)
448 Returns:
449 Self for method chaining
450 """
451 if isinstance(order, str):
452 order = SortOrder.ASC if order.lower() == "asc" else SortOrder.DESC
454 self.sort_specs.append(SortSpec(field=field, order=order))
455 return self
457 def sort(self, field: str, order: str | SortOrder = "asc") -> Query:
458 """Add sorting (fluent interface)."""
459 return self.sort_by(field, order)
461 def set_limit(self, limit: int) -> Query:
462 """Set the result limit (fluent interface).
464 Args:
465 limit: Maximum number of results
467 Returns:
468 Self for method chaining
469 """
470 self.limit_value = limit
471 return self
473 def limit(self, value: int) -> Query:
474 """Set limit (fluent interface)."""
475 return self.set_limit(value)
477 def set_offset(self, offset: int) -> Query:
478 """Set the result offset (fluent interface).
480 Args:
481 offset: Number of results to skip
483 Returns:
484 Self for method chaining
485 """
486 self.offset_value = offset
487 return self
489 def offset(self, value: int) -> Query:
490 """Set offset (fluent interface)."""
491 return self.set_offset(value)
493 def select(self, *fields: str) -> Query:
494 """Set field projection (fluent interface).
496 Args:
497 fields: Field names to include in results
499 Returns:
500 Self for method chaining
501 """
502 self.fields = list(fields) if fields else None
503 return self
505 def clear_filters(self) -> Query:
506 """Clear all filters (fluent interface)."""
507 self.filters = []
508 return self
510 def clear_sort(self) -> Query:
511 """Clear all sort specifications (fluent interface)."""
512 self.sort_specs = []
513 return self
515 def similar_to(
516 self,
517 vector: np.ndarray | list[float],
518 field: str = "embedding",
519 k: int = 10,
520 metric: DistanceMetric | str = "cosine",
521 include_source: bool = True,
522 score_threshold: float | None = None,
523 ) -> Query:
524 """Add vector similarity search to the query.
526 This method sets up a vector similarity search that will find the k most
527 similar vectors to the provided query vector.
529 Args:
530 vector: Query vector to search for similar vectors
531 field: Vector field name to search (default: "embedding")
532 k: Number of results to return (default: 10)
533 metric: Distance metric to use (default: "cosine")
534 include_source: Whether to include source text in results (default: True)
535 score_threshold: Minimum similarity score threshold (optional)
537 Returns:
538 Self for method chaining
539 """
540 self.vector_query = VectorQuery(
541 vector=vector,
542 field_name=field,
543 k=k,
544 metric=metric,
545 include_source=include_source,
546 score_threshold=score_threshold,
547 )
548 # Always update limit to match k
549 self.limit_value = k
550 return self
552 def near_text(
553 self,
554 text: str,
555 embedding_fn: Callable[[str], np.ndarray],
556 field: str = "embedding",
557 k: int = 10,
558 metric: DistanceMetric | str = "cosine",
559 include_source: bool = True,
560 score_threshold: float | None = None,
561 ) -> Query:
562 """Add text-based vector similarity search to the query.
564 This is a convenience method that converts text to a vector using the
565 provided embedding function, then performs vector similarity search.
567 Args:
568 text: Text to convert to vector for similarity search
569 embedding_fn: Function to convert text to vector
570 field: Vector field name to search (default: "embedding")
571 k: Number of results to return (default: 10)
572 metric: Distance metric to use (default: "cosine")
573 include_source: Whether to include source text in results (default: True)
574 score_threshold: Minimum similarity score threshold (optional)
576 Returns:
577 Self for method chaining
578 """
579 # Convert text to vector using provided embedding function
580 vector = embedding_fn(text)
581 return self.similar_to(
582 vector=vector,
583 field=field,
584 k=k,
585 metric=metric,
586 include_source=include_source,
587 score_threshold=score_threshold,
588 )
590 def hybrid(
591 self,
592 text_query: str | None = None,
593 vector: np.ndarray | list[float] | None = None,
594 text_field: str = "content",
595 vector_field: str = "embedding",
596 alpha: float = 0.5,
597 k: int = 10,
598 metric: DistanceMetric | str = "cosine",
599 ) -> Query:
600 """Create a hybrid query combining text and vector search.
602 This method combines traditional text search with vector similarity search,
603 allowing for more nuanced queries that leverage both exact text matching
604 and semantic similarity.
606 Args:
607 text_query: Text to search for (optional)
608 vector: Vector for similarity search (optional)
609 text_field: Field for text search (default: "content")
610 vector_field: Field for vector search (default: "embedding")
611 alpha: Weight balance between text (0.0) and vector (1.0) search (default: 0.5)
612 k: Number of results to return (default: 10)
613 metric: Distance metric for vector search (default: "cosine")
615 Returns:
616 Self for method chaining
618 Note:
619 - alpha=0.0 gives full weight to text search
620 - alpha=1.0 gives full weight to vector search
621 - alpha=0.5 gives equal weight to both
622 """
623 # Add text filter if provided
624 if text_query:
625 self.filter(text_field, Operator.LIKE, f"%{text_query}%")
627 # Add vector search if provided
628 if vector is not None:
629 self.vector_query = VectorQuery(
630 vector=vector,
631 field_name=vector_field,
632 k=k,
633 metric=metric,
634 include_source=True,
635 )
636 # Store alpha in vector query metadata for backend to use
637 self.vector_query.metadata = {"hybrid_alpha": alpha}
639 # Set limit if not already set
640 if self.limit_value is None:
641 self.limit_value = k
643 return self
645 def with_reranking(self, rerank_k: int | None = None) -> Query:
646 """Enable result reranking for vector queries.
648 Args:
649 rerank_k: Number of results to rerank (default: 2*k from vector query)
651 Returns:
652 Self for method chaining
653 """
654 if self.vector_query:
655 self.vector_query.rerank = True
656 self.vector_query.rerank_k = rerank_k or (self.vector_query.k * 2)
657 return self
659 def clear_vector(self) -> Query:
660 """Clear vector search from the query (fluent interface)."""
661 self.vector_query = None
662 return self
664 def to_dict(self) -> dict[str, Any]:
665 """Convert query to dictionary representation."""
666 result = {
667 "filters": [f.to_dict() for f in self.filters],
668 "sort": [s.to_dict() for s in self.sort_specs],
669 }
670 if self.limit_value is not None:
671 result["limit"] = self.limit_value
672 if self.offset_value is not None:
673 result["offset"] = self.offset_value
674 if self.fields is not None:
675 result["fields"] = self.fields
676 if self.vector_query is not None:
677 result["vector_query"] = self.vector_query.to_dict()
678 return result
680 @classmethod
681 def from_dict(cls, data: dict[str, Any]) -> Query:
682 """Create query from dictionary representation."""
683 query = cls()
685 for filter_data in data.get("filters", []):
686 query.filters.append(Filter.from_dict(filter_data))
688 for sort_data in data.get("sort", []):
689 query.sort_specs.append(SortSpec.from_dict(sort_data))
691 query.limit_value = data.get("limit")
692 query.offset_value = data.get("offset")
693 query.fields = data.get("fields")
695 if "vector_query" in data:
696 query.vector_query = VectorQuery.from_dict(data["vector_query"])
698 return query
700 def copy(self) -> Query:
701 """Create a copy of the query."""
702 import copy
704 return Query(
705 filters=copy.deepcopy(self.filters),
706 sort_specs=copy.deepcopy(self.sort_specs),
707 limit_value=self.limit_value,
708 offset_value=self.offset_value,
709 fields=self.fields.copy() if self.fields else None,
710 vector_query=copy.deepcopy(self.vector_query) if self.vector_query else None,
711 )
713 def or_(self, *filters: Filter | Query) -> ComplexQuery:
714 """Create a ComplexQuery with OR logic.
716 The current query's filters become an AND group, combined with OR conditions.
717 Example: Query with filters [A, B] calling or_(C, D) creates: (A AND B) AND (C OR D)
719 Args:
720 filters: Filter objects or Query objects to OR together
722 Returns:
723 ComplexQuery with OR logic
724 """
725 from .query_logic import (
726 ComplexQuery,
727 Condition,
728 FilterCondition,
729 LogicCondition,
730 LogicOperator,
731 )
733 # Build OR conditions from the arguments
734 or_conditions: list[Condition] = []
735 for item in filters:
736 if isinstance(item, Filter):
737 or_conditions.append(FilterCondition(item))
738 elif isinstance(item, Query):
739 if len(item.filters) == 1:
740 or_conditions.append(FilterCondition(item.filters[0]))
741 elif item.filters:
742 and_cond = LogicCondition(operator=LogicOperator.AND)
743 for f in item.filters:
744 and_cond.conditions.append(FilterCondition(f))
745 or_conditions.append(and_cond)
747 # Create the OR condition group
748 or_group = None
749 if or_conditions:
750 if len(or_conditions) == 1:
751 or_group = or_conditions[0]
752 else:
753 or_group = LogicCondition(
754 operator=LogicOperator.OR,
755 conditions=or_conditions
756 )
758 # Combine with existing filters (if any) using AND
759 if self.filters:
760 # Create AND condition for existing filters
761 if len(self.filters) == 1:
762 existing = FilterCondition(self.filters[0])
763 else:
764 existing = LogicCondition(operator=LogicOperator.AND)
765 for f in self.filters:
766 existing.conditions.append(FilterCondition(f))
768 # Combine existing AND new OR group with AND
769 if or_group:
770 root_condition = LogicCondition(
771 operator=LogicOperator.AND,
772 conditions=[existing, or_group]
773 )
774 else:
775 root_condition = existing
776 else:
777 # No existing filters, just use OR group
778 root_condition = or_group
780 return ComplexQuery(
781 condition=root_condition,
782 sort_specs=self.sort_specs.copy(),
783 limit_value=self.limit_value,
784 offset_value=self.offset_value,
785 fields=self.fields.copy() if self.fields else None
786 )
788 def and_(self, *filters: Filter | Query) -> Query:
789 """Add more filters with AND logic (convenience method).
791 Args:
792 filters: Filter objects or Query objects to AND together
794 Returns:
795 Self for chaining
796 """
797 for item in filters:
798 if isinstance(item, Filter):
799 self.filters.append(item)
800 elif isinstance(item, Query):
801 self.filters.extend(item.filters)
802 return self
804 def not_(self, filter: Filter) -> ComplexQuery:
805 """Create a ComplexQuery with NOT logic.
807 Args:
808 filter: Filter to negate
810 Returns:
811 ComplexQuery with NOT logic
812 """
813 from .query_logic import (
814 ComplexQuery,
815 Condition,
816 FilterCondition,
817 LogicCondition,
818 LogicOperator,
819 )
821 # Current filters as AND
822 conditions: list[Condition] = []
823 if self.filters:
824 if len(self.filters) == 1:
825 conditions.append(FilterCondition(self.filters[0]))
826 else:
827 and_cond = LogicCondition(operator=LogicOperator.AND)
828 for f in self.filters:
829 and_cond.conditions.append(FilterCondition(f))
830 conditions.append(and_cond)
832 # Add NOT condition
833 not_cond = LogicCondition(
834 operator=LogicOperator.NOT,
835 conditions=[FilterCondition(filter)]
836 )
837 conditions.append(not_cond)
839 # Create root condition
840 if len(conditions) == 1:
841 root_condition = conditions[0]
842 else:
843 root_condition = LogicCondition(
844 operator=LogicOperator.AND,
845 conditions=conditions
846 )
848 return ComplexQuery(
849 condition=root_condition,
850 sort_specs=self.sort_specs.copy(),
851 limit_value=self.limit_value,
852 offset_value=self.offset_value,
853 fields=self.fields.copy() if self.fields else None
854 )