fix: create schema and table on add_sink

singer_sdk/__init__.py

@@ -4,13 +4,8 @@
 from singer_sdk.mapper_base import InlineMapper
 from singer_sdk.plugin_base import PluginBase
 from singer_sdk.sinks import BatchSink, RecordSink, Sink, SQLSink
-from singer_sdk.streams import (
-    GraphQLStream,
-    RESTStream,
-    SQLConnector,
-    SQLStream,
-    Stream,
-)
+from singer_sdk.sql import SQLConnector
+from singer_sdk.streams import GraphQLStream, RESTStream, SQLStream, Stream
 from singer_sdk.tap_base import SQLTap, Tap
 from singer_sdk.target_base import SQLTarget, Target
 

singer_sdk/sinks/core.py

@@ -423,6 +423,14 @@ def activate_version(self, new_version: int) -> None:
             "Ignoring."
         )
 
+    def setup(self) -> None:
+        """Perform any setup actions at the beginning of a Stream.
+
+        Setup is executed once per Sink instance, after instantiation. If a Schema
+        change is detected, a new Sink is instantiated and this method is called again.
+        """
+        pass
+
     def clean_up(self) -> None:
         """Perform any clean up actions required at end of a stream.
 

singer_sdk/sinks/sql.py

@@ -10,7 +10,7 @@
 
 from singer_sdk.plugin_base import PluginBase
 from singer_sdk.sinks.batch import BatchSink
-from singer_sdk.streams.sql import SQLConnector
+from singer_sdk.sql import SQLConnector
 
 
 class SQLSink(BatchSink):
@@ -38,11 +38,7 @@ def __init__(
             connector: Optional connector to reuse.
         """
         self._connector: SQLConnector
-        if connector:
-            self._connector = connector
-        else:
-            self._connector = self.connector_class(dict(target.config))
-
+        self._connector = connector or self.connector_class(dict(target.config))
         super().__init__(target, stream_name, schema, key_properties)
 
     @property
@@ -65,36 +61,76 @@ def connection(self) -> sqlalchemy.engine.Connection:
 
     @property
     def table_name(self) -> str:
-        """Returns the table name, with no schema or database part.
+        """Return the table name, with no schema or database part.
 
         Returns:
             The target table name.
         """
         parts = self.stream_name.split("-")
-
-        if len(parts) == 1:
-            return self.stream_name
-        else:
-            return parts[-1]
+        return self.stream_name if len(parts) == 1 else parts[-1]
 
     @property
     def schema_name(self) -> Optional[str]:
-        """Returns the schema name or `None` if using names with no schema part.
+        """Return the schema name or `None` if using names with no schema part.
 
         Returns:
             The target schema name.
         """
-        return None  # Assumes single-schema target context.
+        parts = self.stream_name.split("-")
+        if len(parts) == 2:
+            return parts[0]
+        if len(parts) == 3:
+            return parts[1]
+        return None
 
     @property
     def database_name(self) -> Optional[str]:
-        """Returns the DB name or `None` if using names with no database part.
+        """Return the DB name or `None` if using names with no database part.
 
         Returns:
             The target database name.
         """
         return None  # Assumes single-DB target context.
 
+    @property
+    def full_table_name(self) -> str:
+        """Return the fully qualified table name.
+
+        Returns:
+            The fully qualified table name.
+        """
+        return self.connector.get_fully_qualified_name(
+            table_name=self.table_name,
+            schema_name=self.schema_name,
+            db_name=self.database_name,
+        )
+
+    @property
+    def full_schema_name(self) -> str:
+        """Return the fully qualified schema name.
+
+        Returns:
+            The fully qualified schema name.
+        """
+        return self.connector.get_fully_qualified_name(
+            schema_name=self.schema_name, db_name=self.database_name
+        )
+
+    def setup(self) -> None:
+        """Set up Sink.
+
+        This method is called on Sink creation, and creates the required Schema and
+        Table entities in the target database.
+        """
+        if self.schema_name:
+            self.connector.prepare_schema(self.schema_name)
+        self.connector.prepare_table(
+            full_table_name=self.full_table_name,
+            schema=self.schema,
+            primary_keys=self.key_properties,
+            as_temp_table=False,
+        )
+
     def process_batch(self, context: dict) -> None:
         """Process a batch with the given batch context.
 
@@ -106,31 +142,12 @@ def process_batch(self, context: dict) -> None:
         """
         # If duplicates are merged, these can be tracked via
         # :meth:`~singer_sdk.Sink.tally_duplicate_merged()`.
-        self.connector.prepare_table(
-            full_table_name=self.full_table_name,
-            schema=self.schema,
-            primary_keys=self.key_properties,
-            as_temp_table=False,
-        )
         self.bulk_insert_records(
             full_table_name=self.full_table_name,
             schema=self.schema,
             records=context["records"],
         )
 
-    @property
-    def full_table_name(self) -> str:
-        """Gives the fully qualified table name.
-
-        Returns:
-            The fully qualified table name.
-        """
-        return self.connector.get_fully_qualified_name(
-            self.table_name,
-            self.schema_name,
-            self.database_name,
-        )
-
     def create_table_with_records(
         self,
         full_table_name: Optional[str],
@@ -154,6 +171,8 @@ def create_table_with_records(
         if primary_keys is None:
             primary_keys = self.key_properties
         partition_keys = partition_keys or None
+        # TODO: determine if this call to `prepare_table` is necessary
+        # (in addition to in `setup` above)
         self.connector.prepare_table(
             full_table_name=full_table_name,
             primary_keys=primary_keys,

singer_sdk/sql/__init__.py

@@ -0,0 +1,4 @@
+"""Module for helpers common to SQL streams/sinks."""
+from .connector import SQLConnector
+
+__all__ = ["SQLConnector"]