Additional Changes to "Escape quotes in identifiers (#771)" (#811)

* Escape quotes in identifiers

* Add more tests to confirm that escaping quotes in identifiers works and remove FIXME placeholders

---------

Co-authored-by: Alexander Malyga <[email protected]>

Author: jereze

pypika/queries.py

@@ -98,7 +98,6 @@ def __getattr__(self, item: str) -> "Table":
         return Table(item, schema=self)
 
     def get_sql(self, quote_char: Optional[str] = None, **kwargs: Any) -> str:
-        # FIXME escape
         schema_sql = format_quotes(self._name, quote_char)
 
         if self._parent is not None:
@@ -150,7 +149,6 @@ def get_table_name(self) -> str:
 
     def get_sql(self, **kwargs: Any) -> str:
         quote_char = kwargs.get("quote_char")
-        # FIXME escape
         table_sql = format_quotes(self._table_name, quote_char)
 
         if self._schema is not None:

pypika/terms.py

@@ -442,15 +442,13 @@ def get_value_sql(self, **kwargs: Any) -> str:
     def get_formatted_value(cls, value: Any, **kwargs):
         quote_char = kwargs.get("secondary_quote_char") or ""
 
-        # FIXME escape values
         if isinstance(value, Term):
             return value.get_sql(**kwargs)
         if isinstance(value, Enum):
             return cls.get_formatted_value(value.value, **kwargs)
         if isinstance(value, (date, datetime, time)):
             return cls.get_formatted_value(value.isoformat(), **kwargs)
         if isinstance(value, str):
-            value = value.replace(quote_char, quote_char * 2)
             return format_quotes(value, quote_char)
         if isinstance(value, bool):
             return str.lower(str(value))
@@ -956,7 +954,6 @@ def __init__(self, container, alias=None):
         self._is_negated = False
 
     def get_sql(self, **kwargs):
-        # FIXME escape
         return "{not_}EXISTS {container}".format(
             container=self.container.get_sql(**kwargs), not_='NOT ' if self._is_negated else ''
         )
@@ -1000,7 +997,6 @@ def replace_table(self, current_table: Optional["Table"], new_table: Optional["T
         self.term = self.term.replace_table(current_table, new_table)
 
     def get_sql(self, **kwargs: Any) -> str:
-        # FIXME escape
         sql = "{term} BETWEEN {start} AND {end}".format(
             term=self.term.get_sql(**kwargs),
             start=self.start.get_sql(**kwargs),

pypika/tests/test_criterions.py

@@ -731,6 +731,16 @@ def test_not_exists(self):
             'SELECT "t1"."field1" FROM "def" "t1" WHERE NOT EXISTS (SELECT "t2"."field2" FROM "abc" "t2")', str(q1)
         )
 
+    def test_exists_with_double_quotes(self):
+        t3 = Table('abc"', alias='t3"')
+        q3 = QueryBuilder().from_(t3).select(t3.field2)
+        t1 = Table("def", alias="t1")
+        q1 = QueryBuilder().from_(t1).where(ExistsCriterion(q3)).select(t1.field1)
+
+        self.assertEqual(
+            'SELECT "t1"."field1" FROM "def" "t1" WHERE EXISTS (SELECT "t3"""."field2" FROM "abc""" "t3""")', str(q1)
+        )
+
 
 class ComplexCriterionTests(unittest.TestCase):
     table_abc, table_efg = Table("abc", alias="cx0"), Table("efg", alias="cx1")
@@ -798,6 +808,11 @@ def test__between_and_field(self):
         self.assertEqual('"foo" BETWEEN 0 AND 1 AND "bool_field"', str(c1 & c2))
         self.assertEqual('"bool_field" AND "foo" BETWEEN 0 AND 1', str(c2 & c1))
 
+    def test__between_with_quotes(self):
+        c = Field('foo"\'').between("a'", "c'")
+
+        self.assertEqual('"foo""\'" BETWEEN \'a\'\'\' AND \'c\'\'\'', str(c))
+
 
 class FieldsAsCriterionTests(unittest.TestCase):
     def test__field_and_field(self):

pypika/tests/test_selects.py

@@ -49,6 +49,11 @@ def test_select_no_with_alias_from(self):
 
         self.assertEqual('SELECT 1 "test"', str(q))
 
+    def test_select_literal_with_alias_with_quotes(self):
+        q = Query.select(ValueWrapper("contains'\"quotes", "contains'\"quotes"))
+
+        self.assertEqual('SELECT \'contains\'\'"quotes\' "contains\'""quotes"', str(q))
+
     def test_select_no_from_with_field_raises_exception(self):
         with self.assertRaises(QueryException):
             Query.select("asdf")
@@ -73,6 +78,11 @@ def test_select__table_schema_with_multiple_levels_as_list(self):
 
         self.assertEqual('SELECT * FROM "schema1"."schema2"."abc"', str(q))
 
+    def test_select__table_schema_escape_double_quote(self):
+        q = Query.from_(Table("abc", 'schema_with_double_quote"')).select("*")
+
+        self.assertEqual('SELECT * FROM "schema_with_double_quote"""."abc"', str(q))
+
     def test_select__star__replacement(self):
         q = Query.from_("abc").select("foo").select("*")
 

pypika/tests/test_tables.py

@@ -13,11 +13,33 @@ def test_table_sql(self):
 
         self.assertEqual('"test_table"', str(table))
 
+    def test_table_sql_with_double_quote(self):
+        table = Table('test_table_with_double_quote"')
+
+        self.assertEqual('"test_table_with_double_quote"""', str(table))
+
+    def test_table_sql_with_several_double_quotes(self):
+        table = Table('test"table""with"""double"quotes')
+
+        self.assertEqual('"test""table""""with""""""double""quotes"', str(table))
+
+    def test_table_sql_with_single_quote(self):
+        table = Table("test_table_with_single_quote'")
+
+        self.assertEqual('"test_table_with_single_quote\'"', str(table))
+
     def test_table_with_alias(self):
         table = Table("test_table").as_("my_table")
 
         self.assertEqual('"test_table" "my_table"', table.get_sql(with_alias=True, quote_char='"'))
 
+    def test_table_with_alias_with_double_quote(self):
+        table = Table('test_table_with_double_quote"').as_("my_alias\"")
+
+        self.assertEqual(
+            '"test_table_with_double_quote""" "my_alias"""', table.get_sql(with_alias=True, quote_char='"')
+        )
+
     def test_schema_table_attr(self):
         table = Schema("x_schema").test_table
 

pypika/tests/test_terms.py

@@ -80,3 +80,25 @@ def test_passes_kwargs_to_field_get_sql(self):
             'FROM "customers" JOIN "accounts" ON "customers"."account_id"="accounts"."account_id"',
             query.get_sql(with_namespace=True),
         )
+
+
+class IdentifierEscapingTests(TestCase):
+    def test_escape_identifier_quotes(self):
+        customers = Table('customers"')
+        customer_id = getattr(customers, '"id')
+        email = getattr(customers, 'email"').as_('customer_email"')
+
+        query = (
+            Query.from_(customers)
+            .select(customer_id, email)
+            .where(customer_id == "abc")
+            .where(email == "[email protected]")
+            .orderby(email, customer_id)
+        )
+
+        self.assertEqual(
+            'SELECT """id","email""" "customer_email""" '
+            'FROM "customers""" WHERE """id"=\'abc\' AND "email"""=\'[email protected]\' '
+            'ORDER BY "customer_email""","""id"',
+            query.get_sql(),
+        )

pypika/utils.py

@@ -120,6 +120,9 @@ def resolve_is_aggregate(values: List[Optional[bool]]) -> Optional[bool]:
 
 
 def format_quotes(value: Any, quote_char: Optional[str]) -> str:
+    if quote_char:
+        value = value.replace(quote_char, quote_char * 2)
+
     return "{quote}{value}{quote}".format(value=value, quote=quote_char or "")