diff --git a/packages/google-cloud-spanner/google/cloud/spanner_dbapi/parse_utils.py b/packages/google-cloud-spanner/google/cloud/spanner_dbapi/parse_utils.py index c00aaf245ee2..559793d990a8 100644 --- a/packages/google-cloud-spanner/google/cloud/spanner_dbapi/parse_utils.py +++ b/packages/google-cloud-spanner/google/cloud/spanner_dbapi/parse_utils.py @@ -51,6 +51,9 @@ JsonObject: spanner.param_types.JSON, } +# Note: SPANNER_RESERVED_KEYWORDS is retained for backwards compatibility with +# downstream packages (e.g. sqlalchemy-spanner). The Spanner DB-API driver quotes +# identifiers unconditionally and does not use this list internally. SPANNER_RESERVED_KEYWORDS = { "ALL", "AND", @@ -89,6 +92,7 @@ "FOR", "FROM", "FULL", + "GRAPH_TABLE", "GROUP", "GROUPING", "GROUPS", @@ -124,6 +128,7 @@ "PARTITION", "PRECEDING", "PROTO", + "QUALIFY", "RANGE", "RECURSIVE", "RESPECT", @@ -383,37 +388,38 @@ def ensure_where_clause(sql): _QUOTED_IDENTIFIER_RE = re.compile(r"\A`(\\.|[^`\\])*`\Z", re.DOTALL) +_QUALIFIED_PART_RE = re.compile(r"`(?:\\.|[^`\\])*`|[^.]+") +_CONTROL_CHAR_RE = re.compile(r"[\x00-\x1f\x7f]") -def escape_name(name): - """ - Apply backticks to the name if it is not a valid regular ASCII identifier, - or if it is a Cloud Spanner's reserved keyword. +def escape_name(name: str) -> str: + """Apply GoogleSQL backticks to the identifier and escape embedded delimiters. :type name: str - :param name: Name to escape. + :param name: Identifier name to escape. :rtype: str - :returns: Name escaped if it has to be escaped. + :returns: Escaped and quoted identifier. + :raises TypeError: if the name is not a string. + :raises ValueError: if the name contains ASCII control characters or is + an invalid qualified identifier. """ + if not isinstance(name, str): + raise TypeError(f"Identifier must be a string, got {type(name).__name__}") if not name: return name + if _CONTROL_CHAR_RE.search(name): + raise ValueError(f"Invalid identifier: {name!r}") + if _QUOTED_IDENTIFIER_RE.match(name): return name if "." in name: - parts = name.split(".") + parts = _QUALIFIED_PART_RE.findall(name) + if not parts or ".".join(parts) != name: + raise ValueError(f"Invalid qualified identifier: {name!r}") return ".".join(escape_name(part) for part in parts) - is_valid_regular_identifier = ( - (name[0].isalpha() or name[0] == "_") - and all(c.isalnum() or c == "_" for c in name) - and name.isascii() - ) - - if not is_valid_regular_identifier or name.upper() in SPANNER_RESERVED_KEYWORDS: - escaped = name.replace("\\", "\\\\").replace("`", "\\`") - return f"`{escaped}`" - - return name + escaped = name.replace("\\", "\\\\").replace("`", "\\`") + return f"`{escaped}`" diff --git a/packages/google-cloud-spanner/tests/unit/spanner_dbapi/test_parse_utils.py b/packages/google-cloud-spanner/tests/unit/spanner_dbapi/test_parse_utils.py index 71c4bb4d7fda..2bd2570de92b 100644 --- a/packages/google-cloud-spanner/tests/unit/spanner_dbapi/test_parse_utils.py +++ b/packages/google-cloud-spanner/tests/unit/spanner_dbapi/test_parse_utils.py @@ -15,7 +15,11 @@ import sys import unittest -from google.cloud.spanner_dbapi.parse_utils import classify_statement +from google.cloud.spanner_dbapi.parse_utils import ( + SPANNER_RESERVED_KEYWORDS, + classify_statement, + escape_name, +) from google.cloud.spanner_dbapi.parsed_statement import ( ClientSideStatementType, ParsedStatement, @@ -501,27 +505,32 @@ def test_ensure_where_clause(self): @unittest.skipIf(skip_condition, skip_message) def test_escape_name(self): - from google.cloud.spanner_dbapi.parse_utils import escape_name - cases = ( ("SELECT", "`SELECT`"), ("dashed-value", "`dashed-value`"), ("with space", "`with space`"), - ("name", "name"), + ("name", "`name`"), ("", ""), + ("``", "``"), ("col`; DROP TABLE t; -- x", "`col\\`; DROP TABLE t; -- x`"), ("table`name", "`table\\`name`"), ("`", "`\\``"), ("col/*comment*/name", "`col/*comment*/name`"), ("123column", "`123column`"), ("col;select", "`col;select`"), - ("col\nname", "`col\nname`"), ("test\\", "`test\\\\`"), - ("my_schema.my_table", "my_schema.my_table"), + ("my_schema.my_table", "`my_schema`.`my_table`"), ("my-schema.my-table", "`my-schema`.`my-table`"), + ("catalog.schema.table.column", "`catalog`.`schema`.`table`.`column`"), ("`my_table`", "`my_table`"), ("`my_schema`.`my_table`", "`my_schema`.`my_table`"), ("`table.with.dots`", "`table.with.dots`"), + ("`table.with.dots`.col", "`table.with.dots`.`col`"), + ("col.`table.with.dots`", "`col`.`table.with.dots`"), + ("`my.schema`.`my.table`", "`my.schema`.`my.table`"), + ("café", "`café`"), + ("col_ñ", "`col_ñ`"), + ("`col\\`name`", "`col\\`name`"), ("`col\\`; DROP TABLE users; --`", "`col\\`; DROP TABLE users; --`"), ( "`col\\\\`; DROP TABLE users; --`", @@ -532,3 +541,52 @@ def test_escape_name(self): with self.subTest(name=name): got = escape_name(name) self.assertEqual(got, want) + # Verify idempotency: escaping an already-escaped output returns the same result. + self.assertEqual(escape_name(got), want) + + @unittest.skipIf(skip_condition, skip_message) + def test_escape_name_invalid_control_characters(self): + invalid_identifiers = ( + "table\nname", + "table\rname", + "table\0name", + "table\tname", + "table\x1fname", + "table\x7fname", + "schema\n.table", + "schema.table\0", + ) + for identifier in invalid_identifiers: + with self.subTest(identifier=identifier): + with self.assertRaises(ValueError): + escape_name(identifier) + + @unittest.skipIf(skip_condition, skip_message) + def test_escape_name_invalid_qualified_identifiers(self): + invalid_qualified = ( + ".", + "...", + "table.", + ".table", + "schema..table", + "schema.`unclosed.table.`other", + ) + for identifier in invalid_qualified: + with self.subTest(identifier=identifier): + with self.assertRaises(ValueError): + escape_name(identifier) + + @unittest.skipIf(skip_condition, skip_message) + def test_escape_name_invalid_types(self): + invalid_types = (None, 123, 45.67, b"bytes_name", ["list"]) + for identifier in invalid_types: + with self.subTest(identifier=identifier): + with self.assertRaises(TypeError): + escape_name(identifier) + + @unittest.skipIf(skip_condition, skip_message) + def test_spanner_reserved_keywords_compatibility(self): + # Verify that SPANNER_RESERVED_KEYWORDS is exported with recent additions + # for backwards compatibility with downstream consumers (e.g. sqlalchemy-spanner). + self.assertIn("GRAPH_TABLE", SPANNER_RESERVED_KEYWORDS) + self.assertIn("QUALIFY", SPANNER_RESERVED_KEYWORDS)