Merge pull request #206 from UncoderIO/gis-7011

sql str value manager

Author: alexvolha

uncoder-core/app/translator/core/str_value_manager.py

@@ -182,7 +182,11 @@ class StrValueManager:
     container_spec_symbols_map: ClassVar[dict[type[BaseSpecSymbol], str]] = CONTAINER_SPEC_SYMBOLS_MAP
 
     @staticmethod
-    def from_str_to_container(value: str) -> StrValue:
+    def from_str_to_container(
+        value: str,
+        value_type: str = ValueType.value,  # noqa: ARG004
+        escape_symbol: Optional[str] = None,  # noqa: ARG004
+    ) -> StrValue:
         return StrValue(value=value, split_value=[value])
 
     def from_re_str_to_container(self, value: str) -> StrValue:

uncoder-core/app/translator/core/tokenizer.py

@@ -162,8 +162,7 @@ def search_multi_value(
 
     def _get_field_value_match(self, query: str, operator: str, field_name: str, value_pattern: str) -> re.Match:
         field_value_pattern = self.get_field_value_pattern(operator, field_name, value_pattern)
-        field_value_regex = re.compile(field_value_pattern, re.IGNORECASE)
-        field_value_match = re.match(field_value_regex, query)
+        field_value_match = re.match(field_value_pattern, query, re.IGNORECASE)
         if field_value_match is None:
             raise TokenizerGeneralException(error=f"Value couldn't be found in query part: {query}")
 

uncoder-core/app/translator/platforms/anomali/renders/anomali.py

@@ -17,65 +17,37 @@
 -----------------------------------------------------------------
 """
 from app.translator.const import DEFAULT_VALUE_TYPE
-from app.translator.core.custom_types.values import ValueType
 from app.translator.core.models.platform_details import PlatformDetails
-from app.translator.core.render import BaseFieldValueRender, PlatformQueryRender
+from app.translator.core.render import PlatformQueryRender
 from app.translator.managers import render_manager
 from app.translator.platforms.anomali.const import anomali_query_details
 from app.translator.platforms.anomali.mapping import AnomaliMappings, anomali_query_mappings
-from app.translator.platforms.base.sql.str_value_manager import sql_str_value_manager
+from app.translator.platforms.base.sql.renders.sql import SqlFieldValueRender
 
 
-class AnomaliFieldValueRender(BaseFieldValueRender):
+class AnomaliFieldValueRender(SqlFieldValueRender):
     details: PlatformDetails = anomali_query_details
-    str_value_manager = sql_str_value_manager
-
-    @staticmethod
-    def _wrap_str_value(value: str) -> str:
-        return f"'{value}'"
-
-    def equal_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
-        if isinstance(value, list):
-            return f"({self.or_token.join([self.equal_modifier(field=field, value=v) for v in value])})"
-        return f"{field} = {self._pre_process_value(field, value, wrap_str=True)}"
-
-    def not_equal_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
-        if isinstance(value, list):
-            return f"({self.or_token.join([self.not_equal_modifier(field=field, value=v) for v in value])})"
-        return f"{field} != {self._pre_process_value(field, value, wrap_str=True)}"
-
-    def less_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
-        return f"{field} < {self._pre_process_value(field, value, wrap_str=True)}"
-
-    def less_or_equal_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
-        return f"{field} <= {self._pre_process_value(field, value, wrap_str=True)}"
-
-    def greater_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
-        return f"{field} > {self._pre_process_value(field, value, wrap_str=True)}"
-
-    def greater_or_equal_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
-        return f"{field} >= {self._pre_process_value(field, value, wrap_str=True)}"
 
     def contains_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
         if isinstance(value, list):
             return f"({self.or_token.join(self.contains_modifier(field=field, value=v) for v in value)})"
-        return f"{field} like '%{self._pre_process_value(field, value)}%'"
+
+        value = f"'%{self._pre_process_value(field, value)}%'"
+        return f"{field} like {value}"
 
     def endswith_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
         if isinstance(value, list):
             return f"({self.or_token.join(self.endswith_modifier(field=field, value=v) for v in value)})"
-        return f"{field} like '%{self._pre_process_value(field, value)}'"
+
+        value = f"'%{self._pre_process_value(field, value)}'"
+        return f"{field} like {value}"
 
     def startswith_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
         if isinstance(value, list):
             return f"({self.or_token.join(self.startswith_modifier(field=field, value=v) for v in value)})"
-        return f"{field} like '{self._pre_process_value(field, value)}%'"
 
-    def regex_modifier(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
-        if isinstance(value, list):
-            return f"({self.or_token.join(self.regex_modifier(field=field, value=v) for v in value)})"
-        regex_str = self._pre_process_value(field, value, value_type=ValueType.regex_value, wrap_str=True)
-        return f"regexp_like({field}, {regex_str})"
+        value = f"'{self._pre_process_value(field, value)}%'"
+        return f"{field} like {value}"
 
     def keywords(self, field: str, value: DEFAULT_VALUE_TYPE) -> str:
         return f'message contains "{self._pre_process_value(field, value)}"'

uncoder-core/app/translator/platforms/base/aql/str_value_manager.py

@@ -18,7 +18,7 @@
 """
 
 import copy
-from typing import ClassVar
+from typing import ClassVar, Optional
 
 from app.translator.core.custom_types.values import ValueType
 from app.translator.core.str_value_manager import (
@@ -55,7 +55,12 @@ class AQLStrValueManager(StrValueManager):
         "%": UnboundLenWildCard,
     }
 
-    def from_str_to_container(self, value: str) -> StrValue:
+    def from_str_to_container(
+        self,
+        value: str,
+        value_type: str = ValueType.value,  # noqa: ARG002
+        escape_symbol: Optional[str] = None,  # noqa: ARG002
+    ) -> StrValue:
         split = []
         prev_char = None
         for char in value:

uncoder-core/app/translator/platforms/base/lucene/str_value_manager.py

@@ -17,8 +17,9 @@
 -----------------------------------------------------------------
 """
 
-from typing import ClassVar
+from typing import ClassVar, Optional
 
+from app.translator.core.custom_types.values import ValueType
 from app.translator.core.str_value_manager import (
     BaseSpecSymbol,
     ReAnySymbol,
@@ -68,7 +69,12 @@ class LuceneStrValueManager(StrValueManager):
     }
     re_str_spec_symbols_map = RE_STR_SPEC_SYMBOLS_MAP
 
-    def from_str_to_container(self, value: str) -> StrValue:
+    def from_str_to_container(
+        self,
+        value: str,
+        value_type: str = ValueType.value,  # noqa: ARG002
+        escape_symbol: Optional[str] = None,  # noqa: ARG002
+    ) -> StrValue:
         split = []
         prev_char = None
         for char in value:

uncoder-core/app/translator/platforms/base/spl/renders/spl.py

@@ -34,7 +34,12 @@ def _wrap_str_value(value: str) -> str:
         return f'"{value}"'
 
     def _pre_process_value(
-        self, field: str, value: Union[int, str, StrValue], value_type: str = ValueType.value, wrap_str: bool = False
+        self,
+        field: str,
+        value: Union[bool, int, str, StrValue],
+        value_type: str = ValueType.value,
+        wrap_str: bool = False,
+        wrap_int: bool = False,  # noqa: ARG002
     ) -> Union[int, str]:
         value = super()._pre_process_value(field, value, value_type=value_type, wrap_str=wrap_str)
         return self._wrap_str_value(str(value)) if not isinstance(value, str) else value

uncoder-core/app/translator/platforms/base/spl/str_value_manager.py

@@ -16,8 +16,9 @@
 limitations under the License.
 -----------------------------------------------------------------
 """
-from typing import ClassVar
+from typing import ClassVar, Optional
 
+from app.translator.core.custom_types.values import ValueType
 from app.translator.core.str_value_manager import BaseSpecSymbol, StrValue, StrValueManager, UnboundLenWildCard
 from app.translator.platforms.base.spl.escape_manager import spl_escape_manager
 
@@ -26,7 +27,12 @@ class SplStrValueManager(StrValueManager):
     escape_manager = spl_escape_manager
     str_spec_symbols_map: ClassVar[dict[str, type[BaseSpecSymbol]]] = {"*": UnboundLenWildCard}
 
-    def from_str_to_container(self, value: str) -> StrValue:
+    def from_str_to_container(
+        self,
+        value: str,
+        value_type: str = ValueType.value,  # noqa: ARG002
+        escape_symbol: Optional[str] = None,  # noqa: ARG002
+    ) -> StrValue:
         split = []
         prev_char = None
         for char in value:

uncoder-core/app/translator/platforms/base/sql/custom_types/__init__.py

@@ -0,0 +1,5 @@
+from app.translator.core.custom_types.values import ValueType
+
+
+class SQLValueType(ValueType):
+    like_value = "like_value"

uncoder-core/app/translator/platforms/base/sql/custom_types/values.py

@@ -1,14 +1,15 @@
 from typing import ClassVar
 
-from app.translator.core.custom_types.values import ValueType
 from app.translator.core.escape_manager import EscapeManager
 from app.translator.core.models.escape_details import EscapeDetails
+from app.translator.platforms.base.sql.custom_types.values import SQLValueType
 
 
 class SQLEscapeManager(EscapeManager):
     escape_map: ClassVar[dict[str, list[EscapeDetails]]] = {
-        ValueType.value: [EscapeDetails(pattern=r"(')", escape_symbols=r"'\1")],
-        ValueType.regex_value: [
+        SQLValueType.value: [EscapeDetails(pattern=r"(')", escape_symbols=r"'\1")],
+        SQLValueType.like_value: [EscapeDetails(pattern=r"(['%_\\])", escape_symbols=r"\\\1")],
+        SQLValueType.regex_value: [
             EscapeDetails(pattern=r"([$^*+()\[\]{}|.?\-\\])", escape_symbols=r"\\\1"),
             EscapeDetails(pattern=r"(')", escape_symbols=r"'\1"),
         ],