diff --git a/pypika/queries.py b/pypika/queries.py index 223c3c95..df4da1b6 100644 --- a/pypika/queries.py +++ b/pypika/queries.py @@ -94,7 +94,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: @@ -146,7 +145,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: diff --git a/pypika/terms.py b/pypika/terms.py index b25af5c5..5c9088b8 100644 --- a/pypika/terms.py +++ b/pypika/terms.py @@ -367,7 +367,6 @@ 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): @@ -375,7 +374,6 @@ def get_formatted_value(cls, value: Any, **kwargs): if isinstance(value, date): 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)) @@ -841,7 +839,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 '' ) @@ -885,7 +882,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), diff --git a/pypika/tests/test_criterions.py b/pypika/tests/test_criterions.py index fca2ceee..468fc3d1 100644 --- a/pypika/tests/test_criterions.py +++ b/pypika/tests/test_criterions.py @@ -639,6 +639,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") @@ -706,6 +716,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): diff --git a/pypika/tests/test_selects.py b/pypika/tests/test_selects.py index 1ce04937..1d75b858 100644 --- a/pypika/tests/test_selects.py +++ b/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("*") diff --git a/pypika/tests/test_tables.py b/pypika/tests/test_tables.py index f43e7409..6d88a28f 100644 --- a/pypika/tests/test_tables.py +++ b/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 diff --git a/pypika/tests/test_terms.py b/pypika/tests/test_terms.py index 562e0d28..de5e94f7 100644 --- a/pypika/tests/test_terms.py +++ b/pypika/tests/test_terms.py @@ -73,3 +73,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 == "abc@abc.com") + .orderby(email, customer_id) + ) + + self.assertEqual( + 'SELECT """id","email""" "customer_email""" ' + 'FROM "customers""" WHERE """id"=\'abc\' AND "email"""=\'abc@abc.com\' ' + 'ORDER BY "customer_email""","""id"', + query.get_sql(), + ) diff --git a/pypika/utils.py b/pypika/utils.py index ca3e9c4c..32e69cc9 100644 --- a/pypika/utils.py +++ b/pypika/utils.py @@ -103,6 +103,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 "")