Fix percent escape binary mod

src/drivers/base.py

@@ -37,15 +37,19 @@
 
 
 class ClickHouseIdentifierPreparer(compiler.IdentifierPreparer):
-    def quote_identifier(self, value):
-        # Never quote identifiers.
-        return self._escape_identifier(value)
-
-    def quote(self, ident, force=None):
-        return ident
+    def _escape_identifier(self, value):
+        value = value.replace(self.escape_quote, self.escape_to_quote)
+        return value.replace('%', '%%')
 
 
 class ClickHouseCompiler(compiler.SQLCompiler):
+    def visit_mod_binary(self, binary, operator, **kw):
+        return self.process(binary.left, **kw) + " %% " + \
+            self.process(binary.right, **kw)
+
+    def post_process_text(self, text):
+        return text.replace('%', '%%')
+
     def visit_count_func(self, fn, **kw):
         # count accepts zero arguments.
         return 'count%s' % self.process(fn.clause_expr, **kw)

tests/drivers/test_escaping.py

@@ -1,20 +1,22 @@
 from decimal import Decimal
 from datetime import date
 
-from sqlalchemy import literal
+from sqlalchemy import Column, literal
 
+from src import types, engines, Table
 from src.drivers.escaper import Escaper
 from tests.session import session
 from tests.testcase import BaseTestCase
 
 
 class EscapingTestCase(BaseTestCase):
-    def compile(self, clause, **kwargs):
+    def escaped_compile(self, clause, **kwargs):
         return str(self._compile(clause, **kwargs))
 
     def test_select_escaping(self):
+        query = session.query(literal('\t'))
         self.assertEqual(
-            self.compile(session.query(literal('\t')), literal_binds=True),
+            self.escaped_compile(query, literal_binds=True),
             "SELECT '\t' AS param_1"
         )
 
@@ -36,3 +38,22 @@ def test_escaper(self):
             e.escape('str')
 
         self.assertIn('Unsupported param format', str(ex.exception))
+
+    def test_escape_binary_mod(self):
+        query = session.query(literal(1) % literal(2))
+        self.assertEqual(
+            self.compile(query, literal_binds=True),
+            'SELECT 1 %% 2 AS anon_1'
+        )
+
+        table = Table(
+            't', self.metadata(),
+            Column('x', types.Int32, primary_key=True),
+            engines.Memory()
+        )
+
+        query = session.query(table.c.x % table.c.x)
+        self.assertEqual(
+            self.compile(query, literal_binds=True),
+            'SELECT x %% x AS anon_1 FROM t'
+        )