fix(py): quote every component of a table name for SQLAlchemy engines

pkg-py/src/commons/_backends.py

@@ -107,18 +107,10 @@ def list_tables(self) -> list[str]:
         return list(sqlalchemy.inspect(self._engine).get_table_names())
 
     def quote(self, table_id: TableId) -> str:
+        # Quote every part of the table name. Required because Snowflake converts unquoted
+        # names to upper case, which would cause a table name with lowercase to not be found.
         preparer = self._engine.dialect.identifier_preparer
-        quoted = preparer.quote(table_id.table)
-        if table_id.schema is None:
-            return quoted
-        # quote_schema() is given one component at a time: handed
-        # "ANALYTICS.PUBLIC" it produces one identifier containing a dot.
-        outer = ".".join(
-            preparer.quote_schema(part)
-            for part in (table_id.catalog, table_id.schema)
-            if part is not None
-        )
-        return f"{outer}.{quoted}"
+        return ".".join(preparer.quote_identifier(part) for part in table_id.parts)
 
     def columns(self, table_id: TableId) -> list[dict[str, Any]]:
         inspector = sqlalchemy.inspect(self._engine)

pkg-py/tests/test_data_source_engine.py

@@ -207,13 +207,23 @@ def test_the_engine_backend_quotes_each_component_separately() -> None:
     assert quoted == '"ANALYTICS"."PUBLIC"."ORDERS"'
 
 
-def test_the_engine_backend_still_quotes_a_two_part_name() -> None:
+def test_the_engine_backend_quotes_a_lowercase_name_too() -> None:
     backend = EngineBackend(sqlalchemy.create_engine("sqlite://"))
     assert backend.quote(TableId(table="sales", schema="analytics")) == (
-        "analytics.sales"
+        '"analytics"."sales"'
     )
 
 
+def test_a_lowercase_snowflake_table_keeps_its_case() -> None:
+    # SQLAlchemy treats an all-lowercase name as case-insensitive and leaves
+    # it bare, which Snowflake folds to upper case: a table created as
+    # "survey" would be looked up as SURVEY.
+    pytest.importorskip("snowflake.sqlalchemy")
+    backend = EngineBackend(sqlalchemy.create_engine("snowflake://u@account/db"))
+    quoted = backend.quote(TableId(table="survey", schema="LIVE", catalog="COMMONS_CI"))
+    assert quoted == '"COMMONS_CI"."LIVE"."survey"'
+
+
 def test_an_engine_backend_reports_columns_from_its_inspector(
     engine: sqlalchemy.Engine,
 ) -> None: