fix: address review feedback — FULLTEXT consistency, term escaping, Black formatting

Agent-Logs-Url: https://github.com/redis-developer/sql-redis/sessions/a931f501-2012-4aa2-8d18-2bca615d32be

Co-authored-by: nkanu17 <47899917+nkanu17@users.noreply.github.com>

Author: Copilot

sql_redis/query_builder.py

@@ -51,6 +51,12 @@ class QueryBuilder:
     # Characters that need escaping in TAG values
     TAG_SPECIAL_CHARS = r".,<>{}[]\"':;!@#$%^&*()-+=~"
 
+    # Characters that have special meaning in RediSearch free-text queries
+    # (outside of double-quoted phrases) and must be escaped with a backslash.
+    # Only characters likely to appear accidentally in user data are included;
+    # intentional RediSearch features (~, *, %, ^) are intentionally excluded.
+    TEXT_QUERY_SPECIAL_CHARS = frozenset({"\\", "-", "@", "|", "(", ")"})
+
     @staticmethod
     def _escape_text_value(value: str) -> str:
         """Escape characters that are special inside RediSearch double-quoted phrases.
@@ -62,6 +68,23 @@ def _escape_text_value(value: str) -> str:
         # then escape double quotes.
         return value.replace("\\", "\\\\").replace('"', '\\"')
 
+    @classmethod
+    def _escape_fulltext_term(cls, term: str) -> str:
+        """Escape characters that have special meaning in RediSearch free-text queries.
+
+        Applied to individual terms used outside of double-quoted phrases (e.g.,
+        in parenthesized FULLTEXT expressions) so that user input containing
+        RediSearch operator characters like |, -, (, ), @ does not alter the
+        query semantics or produce syntax errors.
+        """
+        result = []
+        for char in term:
+            if char in cls.TEXT_QUERY_SPECIAL_CHARS:
+                result.append(f"\\{char}")
+            else:
+                result.append(char)
+        return "".join(result)
+
     def build_text_condition(
         self,
         field: str | list[str],
@@ -77,7 +100,11 @@ def build_text_condition(
 
         Args:
             field: Field name or list of field names for multi-field search.
-            operator: One of =, !=, FULLTEXT (or MATCH alias), LIKE, FUZZY.
+            operator: One of =, !=, FULLTEXT, LIKE, FUZZY.
+                - = / !=: exact phrase match, value wrapped in double quotes.
+                - FULLTEXT: tokenized keyword search with stopword filtering.
+                - LIKE: prefix/suffix/infix pattern (SQL % → RediSearch *).
+                - FUZZY: Levenshtein fuzzy match.
             value: The search term or pattern.
             negated: If True, prefix with - for negation.
             fuzzy_level: Levenshtein distance for FUZZY (1, 2, or 3). Default 1.
@@ -118,7 +145,9 @@ def build_text_condition(
             escaped = self._escape_text_value(value)
             search_value = f'"{escaped}"'
         elif " " in value and " OR " not in value:
-            # FULLTEXT/MATCH with multi-word: tokenized search with stopword filtering
+            # FULLTEXT with multi-word: tokenized search with stopword filtering.
+            # Each term is escaped to prevent RediSearch operator characters in
+            # user input from changing query semantics.
             words = value.split()
             removed_stopwords = [
                 w for w in words if w.lower() in REDIS_DEFAULT_STOPWORDS
@@ -137,14 +166,24 @@ def build_text_condition(
                     stacklevel=2,
                 )
 
-            terms = " ".join(filtered_words) if filtered_words else value
-            search_value = f"({terms})"
+            if filtered_words:
+                escaped_terms = " ".join(
+                    self._escape_fulltext_term(w) for w in filtered_words
+                )
+            else:
+                # All words were stopwords; pass them through (escaped) so the
+                # query doesn't become empty. RediSearch will still skip them at
+                # query time, but this avoids a syntax error from an empty clause.
+                escaped_terms = " ".join(self._escape_fulltext_term(w) for w in words)
+            search_value = f"({escaped_terms})"
         elif " OR " in value:
             # OR union within text field: split on ' OR ' and join with |
-            or_terms = [t.strip() for t in value.split(" OR ")]
+            or_terms = [
+                self._escape_fulltext_term(t.strip()) for t in value.split(" OR ")
+            ]
             search_value = f"({'|'.join(or_terms)})"
         else:
-            search_value = value
+            search_value = self._escape_fulltext_term(value)
 
         base = f"{prefix}@{field}:{search_value}"
 

tests/test_query_builder.py

@@ -71,7 +71,7 @@ def test_text_negation(self):
         """TEXT field with NOT: -@field:term."""
         builder = QueryBuilder()
         result = builder.build_text_condition(
-            "title", "MATCH", "refurbished", negated=True
+            "title", "FULLTEXT", "refurbished", negated=True
         )
 
         assert result == "-@title:refurbished"
@@ -83,11 +83,18 @@ def test_text_fuzzy_match(self):
 
         assert result == "@title:%laptap%"
 
+    def test_text_fulltext_special_chars_escaped(self):
+        """FULLTEXT term with RediSearch operator chars is escaped to avoid injection."""
+        builder = QueryBuilder()
+        result = builder.build_text_condition("description", "FULLTEXT", "anti-virus")
+
+        assert result == r"@description:anti\-virus"
+
     def test_text_multi_field(self):
         """TEXT multi-field search: (@field1|field2:term)."""
         builder = QueryBuilder()
         result = builder.build_text_condition(
-            ["title", "description"], "MATCH", "wireless"
+            ["title", "description"], "FULLTEXT", "wireless"
         )
 
         assert result == "(@title|description:wireless)"