Remove default implementation

singer_sdk/connectors/sql.py

@@ -990,41 +990,19 @@ def prepare_table(
     def prepare_primary_key(
         self,
         *,
-        full_table_name: str | FullyQualifiedName,
-        primary_keys: t.Sequence[str],
+        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.
         """
-        _, schema_name, table_name = self.parse_full_table_name(full_table_name)
-        meta = sa.MetaData(schema=schema_name)
-        meta.reflect(bind=self._engine, only=[table_name])
-        table = meta.tables[full_table_name]  # type: ignore[index]
-        current_pk_cols = [col.name for col in table.primary_key.columns]
-
-        # Nothing to do
-        if current_pk_cols == primary_keys:
-            return
-
-        new_pk = sa.PrimaryKeyConstraint(*primary_keys)
-
-        # If table has no primary key, add the provided one
-        if not current_pk_cols:
-            with self._connect() as conn, conn.begin():
-                conn.execute(sa.schema.AddConstraint(new_pk))
-                return
-
-        # Drop the existing primary key
-        with self._connect() as conn, conn.begin():
-            conn.execute(sa.schema.DropConstraint(table.primary_key))
-
-        # Add the new primary key
-        if primary_keys:
-            with self._connect() as conn, conn.begin():
-                conn.execute(sa.schema.AddConstraint(new_pk))
+        self.logger.debug("Primary key adaptation is not implemented")
 
     def prepare_column(
         self,

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"