Repository: meltano/sdk
Pull request #908 (2022-08-12 18:22:32)

refactor(targets): Allow overriding the bulk insert statement in `SQLSink`
Small improvement towards #9

codecov[bot]: # [Codecov](https://codecov.io/gh/meltano/sdk/pull/908?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) Report
> Merging [#908](https://codecov.io/gh/meltano/sdk/pull/908?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) (d5560f9) into [main](https://codecov.io/gh/meltano/sdk/commit/cc7e06dba12641f23c8c40e737d0e36ce75c1a0c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) (cc7e06d) will **increase** coverage by `0.02%`.
> The diff coverage is `100.00%`.

```diff
@@            Coverage Diff             @@
##             main     #908      +/-   ##
==========================================
+ Coverage   80.31%   80.33%   +0.02%     
==========================================
  Files          34       34              
  Lines        3413     3418       +5     
  Branches      679      679              
==========================================
+ Hits         2741     2746       +5     
  Misses        499      499              
  Partials      173      173              
```


| [Impacted Files](https://codecov.io/gh/meltano/sdk/pull/908?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) | Coverage Δ | |
|---|---|---|
| [singer\_sdk/sinks/sql.py](https://codecov.io/gh/meltano/sdk/pull/908/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano#diff-c2luZ2VyX3Nkay9zaW5rcy9zcWwucHk=) | `78.94% <100.00%> (+1.48%)` | :arrow_up: |

:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano)

aaronsteers: Scratch my comment about the PR title. It looks great as is. I was reading from my mobile app and was looking in the wrong place. 

Patch:
diff --git a/singer_sdk/sinks/sql.py b/singer_sdk/sinks/sql.py
index 62972a9d..5faaf2ca 100644
--- a/singer_sdk/sinks/sql.py
+++ b/singer_sdk/sinks/sql.py
@@ -1,5 +1,6 @@
 """Sink classes load data to SQL targets."""
 
+from textwrap import dedent
 from typing import Any, Dict, Iterable, List, Optional, Type
 
 import sqlalchemy
@@ -162,6 +163,31 @@ class SQLSink(BatchSink):
             full_table_name=full_table_name, schema=schema, records=records
         )
 
+    def generate_insert_statement(
+        self,
+        full_table_name: str,
+        schema: dict,
+    ) -> str:
+        """Generate an insert statement for the given records.
+
+        Args:
+            full_table_name: the target table name.
+            schema: the JSON schema for the new table.
+
+        Returns:
+            An insert statement.
+        """
+        property_names = list(schema["properties"].keys())
+        statement = dedent(
+            f"""\
+            INSERT INTO {full_table_name}
+            ({", ".join(property_names)})
+            VALUES ({", ".join([f":{name}" for name in property_names])})
+            """
+        )
+
+        return statement.rstrip()
+
     def bulk_insert_records(
         self,
         full_table_name: str,
@@ -183,15 +209,13 @@ class SQLSink(BatchSink):
         Returns:
             True if table exists, False if not, None if unsure or undetectable.
         """
-        property_names = list(schema["properties"].keys())
-        insert_sql = sqlalchemy.text(
-            f"INSERT INTO {full_table_name} "
-            f"({', '.join([n for n in property_names])})"
-            f" VALUES "
-            f"({', '.join([':' + n for n in property_names])})"
+        insert_sql = self.generate_insert_statement(
+            full_table_name,
+            schema,
         )
+        self.logger.info("Inserting with SQL: %s", insert_sql)
         self.connector.connection.execute(
-            insert_sql,
+            sqlalchemy.text(insert_sql),
             records,
         )
         if isinstance(records, list):


Test patch:
diff --git a/tests/core/test_sqlite.py b/tests/core/test_sqlite.py
index f3ea042c..b85ff7d6 100644
--- a/tests/core/test_sqlite.py
+++ b/tests/core/test_sqlite.py
@@ -4,6 +4,7 @@ import json
 from copy import deepcopy
 from io import StringIO
 from pathlib import Path
+from textwrap import dedent
 from typing import Dict, cast
 from uuid import uuid4
 
@@ -11,7 +12,7 @@ import pytest
 
 from samples.sample_tap_sqlite import SQLiteConnector, SQLiteTap
 from samples.sample_target_csv.csv_target import SampleTargetCSV
-from samples.sample_target_sqlite import SQLiteTarget
+from samples.sample_target_sqlite import SQLiteSink, SQLiteTarget
 from singer_sdk import SQLStream
 from singer_sdk import typing as th
 from singer_sdk.helpers._singer import Catalog, MetadataMapping, StreamMetadata
@@ -431,3 +432,49 @@ def test_sqlite_column_no_morph(sqlite_sample_target: SQLTarget):
     target_sync_test(sqlite_sample_target, input=StringIO(tap_output_a), finalize=True)
     # Int should be inserted as string.
     target_sync_test(sqlite_sample_target, input=StringIO(tap_output_b), finalize=True)
+
+
+@pytest.mark.parametrize(
+    "stream_name,schema,key_properties,expected_dml",
+    [
+        (
+            "test_stream",
+            {
+                "type": "object",
+                "properties": {
+                    "id": {"type": "integer"},
+                    "name": {"type": "string"},
+                },
+            },
+            [],
+            dedent(
+                """\
+                INSERT INTO test_stream
+                (id, name)
+                VALUES (:id, :name)"""
+            ),
+        ),
+    ],
+    ids=[
+        "no_key_properties",
+    ],
+)
+def test_sqlite_generate_insert_statement(
+    sqlite_sample_target: SQLiteTarget,
+    stream_name: str,
+    schema: dict,
+    key_properties: list,
+    expected_dml: str,
+):
+    sink = SQLiteSink(
+        sqlite_sample_target,
+        stream_name=stream_name,
+        schema=schema,
+        key_properties=key_properties,
+    )
+
+    dml = sink.generate_insert_statement(
+        sink.full_table_name,
+        sink.schema,
+    )
+    assert dml == expected_dml