revise base Query class (#41)

Author: k-schnickmann

search_query/ebsco/parser.py

@@ -256,7 +256,7 @@ def parse_query_tree(
                 parent, current_operator = self.append_operator(parent, proximity_node)
 
             elif token.type == TokenTypes.LOGIC_OPERATOR:
-                new_operator_node = Query(
+                new_operator_node = Query.create(
                     value=token.value.upper(),
                     position=token.position,
                     search_field=search_field or field_context,

search_query/pubmed/parser.py

@@ -186,7 +186,7 @@ def _parse_compound_query(self, tokens: list) -> Query:
         query_start_pos = tokens[0].position[0]
         query_end_pos = tokens[-1].position[1]
 
-        return Query(
+        return Query.create(
             value=operator_type,
             search_field=None,
             children=list(children),
@@ -313,7 +313,7 @@ def _parse_operator_node(self, token_nr: int) -> Query:
         if operator.upper() in {"|", "OR"}:
             operator = Operators.OR
 
-        return Query(
+        return Query.create(
             value=operator,
             search_field=None,
             children=children,

search_query/pubmed/translator.py

@@ -2,10 +2,13 @@
 """Pubmed query translator."""
 from search_query.constants import Fields
 from search_query.constants import Operators
+from search_query.constants import PLATFORM
 from search_query.pubmed.constants import generic_search_field_to_syntax_field
 from search_query.pubmed.constants import syntax_str_to_generic_search_field_set
 from search_query.query import Query
 from search_query.query import SearchField
+from search_query.query_or import OrQuery
+from search_query.query_term import Term
 from search_query.translator_base import QueryTranslator
 
 
@@ -35,11 +38,9 @@ def _expand_flat_or_chains(cls, query: "Query") -> bool:
                 if not child.search_field:  # pragma: no cover
                     continue
                 child.search_field.value = Fields.TITLE
-                new_child = Query(
+                new_child = Term(
                     value=child.value,
-                    operator=False,
                     search_field=SearchField(value=Fields.ABSTRACT),
-                    children=None,
                 )
                 query.add_child(new_child)
 
@@ -146,18 +147,16 @@ def _expand_combined_fields(cls, query: Query, search_fields: set) -> None:
         # Note: sorted list for deterministic order of fields
         for search_field in sorted(list(search_fields)):
             query_children.append(
-                Query(
+                Term(
                     value=query.value,
-                    operator=False,
                     search_field=SearchField(value=search_field),
-                    children=None,
                 )
             )
-
-        query.value = Operators.OR
-        query.operator = True
-        query.search_field = None
-        query.children = query_children  # type: ignore
+        query.replace(
+            OrQuery(
+                children=query_children,
+            )
+        )
 
     @classmethod
     def to_generic_syntax(cls, query: "Query") -> "Query":

search_query/query.py

@@ -31,12 +31,16 @@ def __init__(
         position: typing.Optional[typing.Tuple[int, int]] = None,
         platform: str = "generic",
     ) -> None:
+        if type(self) is Query:
+            raise TypeError(
+                "The base Query type cannot be instantiated directly. "
+                "Use Query.create() or the appropriate query subclass."
+            )
         self._value: str = ""
-        self._operator = False
+        self._operator = operator
         self._children: typing.List[Query] = []
         self._search_field = None
 
-        self.operator = operator
         self.value = value
         if isinstance(search_field, str):
             self.search_field = SearchField(search_field)
@@ -62,6 +66,58 @@ def __init__(
         # when queries are created programmatically
         self._validate_platform_constraints()
 
+    @classmethod
+    def create(
+            cls,
+            value: str,
+            *,
+            operator: bool = True,
+            search_field: typing.Optional[SearchField] = None,
+            children: typing.Optional[typing.List[typing.Union[str, Query]]] = None,
+            position: typing.Optional[typing.Tuple[int, int]] = None,
+            platform: str = "generic",
+            distance: int = 0
+    ) -> Query:
+        """Factory method for query creation."""
+        if not operator:
+            from search_query.query_term import Term
+            return Term(
+                value=value,
+                search_field=search_field,
+                position=position,
+                platform=platform
+            )
+
+        args = {
+            "search_field": search_field,
+            "children": children,
+            "position": position,
+            "platform": platform
+        }
+
+        if value == Operators.AND:
+            from search_query.query_and import AndQuery
+            return AndQuery(**args)
+
+        elif value == Operators.OR:
+            from search_query.query_or import OrQuery
+            return OrQuery(**args)
+
+        elif value == Operators.NOT:
+            from search_query.query_not import NotQuery
+            return NotQuery(**args)
+
+        elif value in {Operators.NEAR, Operators.WITHIN}:
+            from search_query.query_near import NEARQuery
+            return NEARQuery(value=value, distance=distance, **args)
+
+        elif value == Operators.RANGE:
+            from search_query.query_range import RangeQuery
+            return RangeQuery(**args)
+
+        else:
+            raise ValueError(f"Invalid operator value: {value}")
+
     def _validate_platform_constraints(self) -> None:
         if self.platform == "deactivated":
             return
@@ -170,15 +226,18 @@ def value(self, v: str) -> None:
         """Set value property."""
         if not isinstance(v, str):
             raise TypeError("value must be a string")
-        if self.operator and v not in [
-            Operators.AND,
-            Operators.OR,
-            Operators.NOT,
-            Operators.NEAR,
-            Operators.WITHIN,
-            Operators.RANGE,
-        ]:
-            raise ValueError(f"Invalid operator value: {v}")
+        if self.operator:
+            if self._value:
+                raise AttributeError("operator value can only be set once")
+            if v not in [
+                Operators.AND,
+                Operators.OR,
+                Operators.NOT,
+                Operators.NEAR,
+                Operators.WITHIN,
+                Operators.RANGE,
+            ]:
+                raise ValueError(f"Invalid operator value: {v}")
         self._value = v
 
     @property
@@ -191,6 +250,8 @@ def operator(self, is_op: bool) -> None:
         """Set operator property."""
         if not isinstance(is_op, bool):
             raise TypeError("operator must be a boolean")
+        if is_op != self._operator:
+            raise AttributeError("operator property can only be set once")
         self._operator = is_op
 
     @property
@@ -252,6 +313,15 @@ def search_field(self, sf: typing.Optional[SearchField]) -> None:
         """Set search field property."""
         self._search_field = copy.deepcopy(sf) if sf else None
 
+    def replace(self, new_query) -> None:
+        if self.get_parent():
+            for index, child in enumerate(self.get_parent().children):
+                if child is self:
+                    self.get_parent().children[index] = new_query
+                    return
+        else:
+            raise RuntimeError("Root node of a query cannot be replaced")
+
     def selects(self, *, record_dict: dict) -> bool:
         """Indicates whether the query selects a given record."""
         # pylint: disable=import-outside-toplevel

search_query/query_not.py

@@ -6,6 +6,7 @@
 from typing import Union
 
 from search_query.constants import Operators
+from search_query.constants import PLATFORM
 from search_query.query import Query
 from search_query.query import SearchField
 from search_query.query_term import Term
@@ -32,7 +33,7 @@ def __init__(
 
         query_children = [
             c if isinstance(c, Query) else Term(value=c) for c in children
-        ]
+        ] if children else None
 
         super().__init__(
             value=Operators.NOT,
@@ -58,10 +59,10 @@ def children(self, children: typing.List[Query]) -> None:
         """Set the children of NOT query, updating parent pointers."""
         # Clear existing children and reset parent links (if necessary)
         self._children.clear()
-        if not isinstance(children, list):
+        if self.platform != "deactivated" and not isinstance(children, list):
             raise TypeError("children must be a list of Query instances or strings")
 
-        if len(children) != 2:
+        if self.platform not in {"deactivated", PLATFORM.WOS} and len(children) != 2:
             raise ValueError("A NOT query must have two children")
 
         # Add each new child using add_child (ensures parent is set)

search_query/wos/parser.py

@@ -18,6 +18,7 @@
 from search_query.query import SearchField
 from search_query.query_near import NEARQuery
 from search_query.query_term import Term
+from search_query.query_not import NotQuery
 from search_query.wos.constants import search_field_general_to_syntax
 from search_query.wos.linter import WOSQueryListLinter
 from search_query.wos.linter import WOSQueryStringLinter
@@ -133,9 +134,7 @@ def parse_query_tree(
 
                 if current_negation:
                     # If the current operator is NOT, wrap the sub_query in a NOT
-                    not_part = Query(
-                        value="NOT",
-                        operator=True,
+                    not_part = NotQuery(
                         children=[sub_query],
                         search_field=search_field,
                         platform="deactivated",
@@ -181,9 +180,7 @@ def parse_query_tree(
             # Handle terms
             elif token.type == TokenTypes.SEARCH_TERM:
                 if current_negation:
-                    not_part = Query(
-                        value="NOT",
-                        operator=True,
+                    not_part = NotQuery(
                         children=[
                             Term(
                                 value=token.value,
@@ -219,7 +216,7 @@ def parse_query_tree(
 
         # Return the operator and children if there is an operator
         return (
-            Query(
+            Query.create(
                 value=current_operator,
                 children=list(children),
                 search_field=search_field,
@@ -257,7 +254,7 @@ def _handle_closing_parenthesis(
                     platform="deactivated",
                     distance=distance,
                 )
-            return Query(
+            return Query.create(
                 value=current_operator,
                 children=children,
                 search_field=search_field,
@@ -398,7 +395,7 @@ def _build_query_from_operator_node(self, tokens: list) -> Query:
 
         assert operator, "[ERROR] No operator found in combining query."
 
-        operator_query = Query(
+        operator_query = Query.create(
             value=operator,
             children=children,
             platform="deactivated",

search_query/wos/translator.py

@@ -6,6 +6,8 @@
 from search_query.constants import Operators
 from search_query.query import Query
 from search_query.query import SearchField
+from search_query.query_or import OrQuery
+from search_query.query_term import Term
 from search_query.translator_base import QueryTranslator
 from search_query.wos.constants import generic_search_field_to_syntax_field
 from search_query.wos.constants import syntax_str_to_generic_search_field_set
@@ -44,18 +46,17 @@ def _expand_combined_fields(cls, query: Query, search_fields: set) -> None:
         # Note: sorted list for deterministic order of fields
         for search_field in sorted(list(search_fields)):
             query_children.append(
-                Query(
+                Term(
                     value=query.value,
-                    operator=False,
                     search_field=SearchField(value=search_field),
-                    children=None,
                 )
             )
 
-        query.value = Operators.OR
-        query.operator = True
-        query.search_field = None
-        query.children = query_children  # type: ignore
+        query.replace(
+            OrQuery(
+                children=query_children,
+            )
+        )
 
     @classmethod
     def combine_equal_search_fields(cls, query: Query) -> None: