refactor: Backwards-compatible identifier quoting in fully qualified …

…names (#2603)

singer_sdk/connectors/sql.py

@@ -52,7 +52,6 @@ def __init__(
         schema: str | None = None,
         database: str | None = None,
         delimiter: str = ".",
-        dialect: sa.engine.Dialect,
     ) -> None:
         """Initialize the fully qualified table name.
 
@@ -61,7 +60,6 @@ def __init__(
             schema: The name of the schema. Defaults to None.
             database: The name of the database. Defaults to None.
             delimiter: The delimiter to use between parts. Defaults to '.'.
-            dialect: The SQLAlchemy dialect to use for quoting.
 
         Raises:
             ValueError: If the fully qualified name could not be generated.
@@ -70,7 +68,6 @@ def __init__(
         self.schema = schema
         self.database = database
         self.delimiter = delimiter
-        self.dialect = dialect
 
         parts = []
         if self.database:
@@ -94,7 +91,7 @@ def __init__(
 
         super().__init__(self.delimiter.join(parts))
 
-    def prepare_part(self, part: str) -> str:
+    def prepare_part(self, part: str) -> str:  # noqa: PLR6301
         """Prepare a part of the fully qualified name.
 
         Args:
@@ -103,7 +100,7 @@ def prepare_part(self, part: str) -> str:
         Returns:
             The prepared part.
         """
-        return self.dialect.identifier_preparer.quote(part)
+        return part
 
 
 class SQLConnector:  # noqa: PLR0904
@@ -322,8 +319,8 @@ def to_sql_type(jsonschema_type: dict) -> sa.types.TypeEngine:
         """
         return th.to_sql_type(jsonschema_type)
 
+    @staticmethod
     def get_fully_qualified_name(
-        self,
         table_name: str | None = None,
         schema_name: str | None = None,
         db_name: str | None = None,
@@ -345,7 +342,6 @@ def get_fully_qualified_name(
             schema=schema_name,
             database=db_name,
             delimiter=delimiter,
-            dialect=self._dialect,
         )
 
     @property

tests/core/test_connector_sql.py

@@ -360,32 +360,35 @@ def create_engine(self) -> Engine:
 
 
 def test_fully_qualified_name():
-    dialect = DefaultDialect()
-
-    fqn = FullyQualifiedName(table="my_table", dialect=dialect)
+    fqn = FullyQualifiedName(table="my_table")
     assert fqn == "my_table"
 
-    fqn = FullyQualifiedName(schema="my_schema", table="my_table", dialect=dialect)
+    fqn = FullyQualifiedName(schema="my_schema", table="my_table")
     assert fqn == "my_schema.my_table"
 
     fqn = FullyQualifiedName(
         database="my_catalog",
         schema="my_schema",
         table="my_table",
-        dialect=dialect,
     )
     assert fqn == "my_catalog.my_schema.my_table"
 
 
 def test_fully_qualified_name_with_quoting():
+    class QuotedFullyQualifiedName(FullyQualifiedName):
+        def __init__(self, *, dialect: sa.engine.Dialect, **kwargs: t.Any):
+            self.dialect = dialect
+            super().__init__(**kwargs)
+
+        def prepare_part(self, part: str) -> str:
+            return self.dialect.identifier_preparer.quote(part)
+
     dialect = DefaultDialect()
 
-    fqn = FullyQualifiedName(table="order", schema="public", dialect=dialect)
+    fqn = QuotedFullyQualifiedName(table="order", schema="public", dialect=dialect)
     assert fqn == 'public."order"'
 
 
 def test_fully_qualified_name_empty_error():
-    dialect = DefaultDialect()
-
     with pytest.raises(ValueError, match="Could not generate fully qualified name"):
-        FullyQualifiedName(dialect=dialect)
+        FullyQualifiedName()