feat: Added SQLConnector.prepare_primary_key for target to implemen…

…t for custom table primary key adaptation (#2730)

* feat: Added `SQLConnector.prepare_primary_key`

* Use `full_table_name` to get table object

* mypy is slightly wrong

* Add built-in test

* Remove default implementation

singer_sdk/connectors/sql.py

@@ -982,6 +982,28 @@ def prepare_table(
                 self.to_sql_type(property_def),
             )
 
+        self.prepare_primary_key(
+            full_table_name=full_table_name,
+            primary_keys=primary_keys,
+        )
+
+    def prepare_primary_key(
+        self,
+        *,
+        full_table_name: str | FullyQualifiedName,  # noqa: ARG002
+        primary_keys: t.Sequence[str],  # noqa: ARG002
+    ) -> None:
+        """Adapt target table primary key to provided schema if possible.
+
+        Implement this method in a subclass to adapt the primary key of the target table
+        to the provided one if possible.
+
+        Args:
+            full_table_name: the target table name.
+            primary_keys: list of key properties.
+        """
+        self.logger.debug("Primary key adaptation is not implemented")
+
     def prepare_column(
         self,
         full_table_name: str | FullyQualifiedName,

singer_sdk/testing/suites.py

@@ -36,6 +36,7 @@
     TargetInvalidSchemaTest,
     TargetNoPrimaryKeys,
     TargetOptionalAttributes,
+    TargetPrimaryKeyUpdates,
     TargetRecordBeforeSchemaTest,
     TargetRecordMissingKeyProperty,
     TargetRecordMissingOptionalFields,
@@ -104,6 +105,7 @@ class TestSuite(t.Generic[T]):
         # TargetMultipleStateMessages,
         TargetNoPrimaryKeys,
         TargetOptionalAttributes,
+        TargetPrimaryKeyUpdates,
         TargetRecordBeforeSchemaTest,
         TargetRecordMissingKeyProperty,
         TargetRecordMissingOptionalFields,

singer_sdk/testing/target_test_streams/pk_updates.singer

@@ -0,0 +1,4 @@
+{"type": "SCHEMA", "stream": "example_stream", "schema": {"properties": {"id": {"type": "integer"}, "name": {"type": "string"}, "email": {"type": "string"}}, "key_properties": ["id"]}}
+{"type": "RECORD", "stream": "example_stream", "record": {"id": 1, "name": "Alice", "email": "[email protected]"}}
+{"type": "SCHEMA", "stream": "example_stream", "schema": {"properties": {"id": {"type": "integer"}, "name": {"type": "string"}, "email": {"type": "string"}}, "key_properties": ["email"]}}
+{"type": "RECORD", "stream": "example_stream", "record": {"id": 2, "name": "Bob", "email": "[email protected]"}}

singer_sdk/testing/target_tests.py

@@ -135,6 +135,12 @@ class TargetSchemaUpdates(TargetFileTestTemplate):
     name = "schema_updates"
 
 
+class TargetPrimaryKeyUpdates(TargetFileTestTemplate):
+    """Test Target handles Primary Key updates."""
+
+    name = "pk_updates"
+
+
 class TargetSpecialCharsInAttributes(TargetFileTestTemplate):
     """Test Target handles special chars in attributes."""
 

tests/core/test_connector_sql.py

@@ -321,9 +321,7 @@ def test_column_rename(self, connector: DuckDBConnector):
         connector.rename_column("test_table", "old_name", "new_name")
 
         with engine.connect() as conn:
-            result = conn.execute(
-                sa.text("SELECT * FROM test_table"),
-            )
+            result = conn.execute(sa.text("SELECT * FROM test_table"))
             assert result.keys() == ["id", "new_name"]
 
     def test_adapt_column_type(self, connector: DuckDBConnector):
@@ -341,9 +339,7 @@ def test_adapt_column_type(self, connector: DuckDBConnector):
         connector._adapt_column_type("test_table", "name", sa.types.String())
 
         with engine.connect() as conn:
-            result = conn.execute(
-                sa.text("SELECT * FROM test_table"),
-            )
+            result = conn.execute(sa.text("SELECT * FROM test_table"))
             assert result.keys() == ["id", "name"]
             assert result.cursor.description[1][1] == "STRING"