From db06328f4ba11962e834e1d229ceb4e2c369a2bc Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Edgar=20Ram=C3=ADrez=20Mondrag=C3=B3n?=
 <edgarrm358@gmail.com>
Date: Tue, 28 Jan 2025 10:56:53 -0600
Subject: [PATCH] fix: Use a SQLAlchemy to generate an insert statement

---
 singer_sdk/connectors/sql.py        |  6 +++-
 singer_sdk/sinks/sql.py             | 43 +++++++++++++++++++----------
 tests/core/sinks/test_sql_sink.py   | 16 ++++++-----
 tests/samples/test_target_sqlite.py |  9 ++----
 4 files changed, 44 insertions(+), 30 deletions(-)

diff --git a/singer_sdk/connectors/sql.py b/singer_sdk/connectors/sql.py
index 1b0f646cc..ee1a141f7 100644
--- a/singer_sdk/connectors/sql.py
+++ b/singer_sdk/connectors/sql.py
@@ -840,7 +840,11 @@ def create_engine(self) -> Engine:
                 pool_pre_ping=True,
             )
 
-    def quote(self, name: str) -> str:
+    @deprecated(
+        "This method is deprecated. Use or override `FullyQualifiedName` instead.",
+        category=SingerSDKDeprecationWarning,
+    )
+    def quote(self, name: str) -> str:  # pragma: no cover
         """Quote a name if it needs quoting, using '.' as a name-part delimiter.
 
         Examples:
diff --git a/singer_sdk/sinks/sql.py b/singer_sdk/sinks/sql.py
index 5bb64afad..855ee351f 100644
--- a/singer_sdk/sinks/sql.py
+++ b/singer_sdk/sinks/sql.py
@@ -4,12 +4,12 @@
 
 import re
 import typing as t
+import warnings
 from collections import defaultdict
 from copy import copy
-from textwrap import dedent
 
 import sqlalchemy as sa
-from sqlalchemy.sql import quoted_name
+from sqlalchemy.sql import insert
 from sqlalchemy.sql.expression import bindparam
 
 from singer_sdk.connectors import SQLConnector
@@ -282,19 +282,26 @@ def generate_insert_statement(
         Returns:
             An insert statement.
         """
-        property_names = list(self.conform_schema(schema)["properties"].keys())
-        column_identifiers = [
-            self.connector.quote(quoted_name(name, quote=True))
-            for name in property_names
-        ]
-        statement = dedent(
-            f"""\
-            INSERT INTO {full_table_name}
-            ({", ".join(column_identifiers)})
-            VALUES ({", ".join([f":{name}" for name in property_names])})
-            """,
+        conformed_schema = self.conform_schema(schema)
+        property_names = list(conformed_schema["properties"])
+
+        _, schema_name, table_name = self.connector.parse_full_table_name(
+            full_table_name
         )
-        return statement.rstrip()
+
+        table = sa.Table(
+            table_name,
+            sa.MetaData(),
+            *[
+                sa.Column(
+                    name, sa.String
+                )  # Assuming all columns are of type String for simplicity  # noqa: E501
+                for name in property_names
+            ],
+            schema=schema_name,
+        )
+
+        return insert(table)
 
     def bulk_insert_records(
         self,
@@ -321,7 +328,13 @@ def bulk_insert_records(
             full_table_name,
             schema,
         )
-        if isinstance(insert_sql, str):
+        if isinstance(insert_sql, str):  # pragma: no cover
+            warnings.warn(
+                "Generating a SQL insert statement as a string is deprecated. "
+                "Please return an SQLAlchemy Executable object instead.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
             insert_sql = sa.text(insert_sql)
 
         conformed_records = [self.conform_record(record) for record in records]
diff --git a/tests/core/sinks/test_sql_sink.py b/tests/core/sinks/test_sql_sink.py
index c20be40f6..d1c921eca 100644
--- a/tests/core/sinks/test_sql_sink.py
+++ b/tests/core/sinks/test_sql_sink.py
@@ -1,9 +1,9 @@
 from __future__ import annotations
 
 import typing as t
-from textwrap import dedent
 
 import pytest
+from sqlalchemy.sql import Insert
 
 from samples.sample_duckdb import DuckDBConnector
 from singer_sdk.sinks.sql import SQLSink
@@ -55,10 +55,12 @@ def sink(self, target: DuckDBTarget, schema: dict) -> DuckDBSink:
 
     def test_generate_insert_statement(self, sink: DuckDBSink, schema: dict):
         """Test that the insert statement is generated correctly."""
-        expected = dedent(
-            """\
-            INSERT INTO foo
-            (id, col_ts, "table")
-            VALUES (:id, :col_ts, :table)"""
+        stmt = sink.generate_insert_statement("foo", schema=schema)
+        assert isinstance(stmt, Insert)
+        assert stmt.table.name == "foo"
+        assert stmt.table.columns.keys() == ["id", "col_ts", "table"]
+
+        # Rendered SQL should look like:
+        assert str(stmt) == (
+            'INSERT INTO foo (id, col_ts, "table") VALUES (:id, :col_ts, :table)'
         )
-        assert sink.generate_insert_statement("foo", schema=schema) == expected
diff --git a/tests/samples/test_target_sqlite.py b/tests/samples/test_target_sqlite.py
index 48e3a8913..d1be34a29 100644
--- a/tests/samples/test_target_sqlite.py
+++ b/tests/samples/test_target_sqlite.py
@@ -646,12 +646,7 @@ def test_record_with_missing_properties(
                 },
             },
             [],
-            dedent(
-                """\
-                INSERT INTO test_stream
-                (id, name, "table")
-                VALUES (:id, :name, :table)""",
-            ),
+            'INSERT INTO test_stream (id, name, "table") VALUES (:id, :name, :table)',
         ),
     ],
     ids=[
@@ -676,7 +671,7 @@ def test_sqlite_generate_insert_statement(
         sink.full_table_name,
         sink.schema,
     )
-    assert dml == expected_dml
+    assert str(dml) == expected_dml
 
 
 def test_hostile_to_sqlite(