feat: Allow SQL targets to customize how they parse <db>-<schema>-<table> and <schema>-<table> stream names explicitly

packages/meltano-target-sqlite/target_sqlite/target.py

@@ -113,6 +113,33 @@ class SQLiteSink(SQLSink[SQLiteConnector]):
 
     connector_class = SQLiteConnector
 
+    # `main` and `temp` are SQLite's own built-in schema names, so a database or schema
+    # segment equal to either is dropped rather than folded in.
+    _default_schemas: t.ClassVar[set[str]] = {"main", "temp"}
+
+    @override
+    def parse_stream_name(self, stream_name: str) -> tuple[str | None, str | None, str]:
+        """Parse a stream name into its database, schema, and table parts.
+
+        SQLite has no `CREATE SCHEMA`/`CREATE DATABASE` and no cross-database
+        addressing, so a `<db>-<schema>-<table>` or `<schema>-<table>` stream name
+        is folded into a single physical table name (`db__schema__table`) instead.
+        `database_name`/`schema_name` are left at their inherited defaults, which
+        then come out `None` since they're derived from this same parsed tuple.
+        """
+        db_name, schema_name, table_name = super().parse_stream_name(stream_name)
+        qualifiers = [
+            part
+            for part in (db_name, schema_name)
+            if part and part.lower() not in self._default_schemas
+        ]
+        return None, None, "__".join([*qualifiers, table_name])
+
+    @override
+    @property
+    def schema_name(self) -> str | None:
+        return None
+
 
 class SQLiteTarget(SQLTarget):
     """The Tap class for SQLite."""

singer_sdk/sql/sink.py

@@ -37,7 +37,7 @@
 _C = t.TypeVar("_C", bound=SQLConnector)
 
 
-class SQLSink(BatchSink, t.Generic[_C]):
+class SQLSink(BatchSink, t.Generic[_C]):  # noqa: PLR0904
     """SQL-type sink type."""
 
     connector_class: type[_C]
@@ -80,11 +80,61 @@ def default_target_schema(self) -> str | None:
         """Default target schema."""
         return self.config.get("default_target_schema", None)  # type: ignore[no-any-return]
 
+    @t.final
+    @functools.cached_property
+    def stream_name_parts(self) -> tuple[str | None, str | None, str]:
+        """Parsed stream name parts (database, schema, table)."""
+        return self.parse_stream_name(self.stream_name)
+
+    def parse_stream_name(  # noqa: PLR6301
+        self,
+        stream_name: str,
+    ) -> tuple[str | None, str | None, str]:
+        """Parse a stream name into its database, schema, and table parts.
+
+        Developers may override this method if their stream naming convention
+        does not follow the traditional pattern: `<table>`, `<schema>-<table>`,
+        or `<db>-<schema>-<table>`.
+
+        Examples:
+            A target that reads `database`/`schema` from config, falling back to
+            the stream name convention, and always upper-cases both. Note that
+            `db_name` is taken only from config, never from a
+            `<db>-<schema>-<table>` stream name:
+
+            .. code-block:: python
+
+                 def parse_stream_name(
+                     self,
+                     stream_name: str,
+                 ) -> tuple[str | None, str | None, str]:
+                     _, schema_name, table_name = super().parse_stream_name(stream_name)
+                     schema_name = schema_name or self.config.get("schema")
+                     db_name = self.config.get("database")
+                     return (
+                         db_name.upper() if db_name else None,
+                         schema_name.upper() if schema_name else None,
+                         table_name,
+                     )
+
+        Args:
+            stream_name: The stream name to parse.
+
+        Returns:
+            A three part tuple (db_name, schema_name, table_name) with any
+            unspecified parts returned as None.
+        """
+        parts = stream_name.split("-")
+        if len(parts) == 1:
+            return None, None, parts[0]
+        if len(parts) == 2:  # noqa: PLR2004
+            return None, parts[0], parts[1]
+        return parts[-3], parts[-2], parts[-1]
+
     @property
     def table_name(self) -> str:
         """Table name, with no schema or database part."""
-        parts = self.stream_name.split("-")
-        table = self.stream_name if len(parts) == 1 else parts[-1]
+        _, _, table = self.stream_name_parts
         return self.conform_name(table, "table")
 
     @property
@@ -97,13 +147,31 @@ def schema_name(self) -> str | None:
         if self.default_target_schema:
             return self.default_target_schema
 
-        parts = self.stream_name.split("-")
-        return self.conform_name(parts[-2], "schema") if len(parts) in {2, 3} else None
+        _, schema, _ = self.stream_name_parts
+        return self.conform_name(schema, "schema") if schema else None
 
     @property
     def database_name(self) -> str | None:
-        """Database name or `None` if using names with no database part."""
-        # Assumes single-DB target context.
+        """Database name or `None` if using names with no database part.
+
+        Assumes single-DB target context by default: the database segment of
+        a `<db>-<schema>-<table>` stream name is ignored. Developers may
+        override this to honor the database segment for targets that support it.
+
+        Examples:
+            Honor the database segment of a `<db>-<schema>-<table>` stream name:
+
+            .. code-block:: python
+
+                 @property
+                 def database_name(self) -> str | None:
+                     db_name, _, _ = self.stream_name_parts
+                     return db_name
+
+        Returns:
+            The database name, or `None` if not applicable.
+        """
+        return None
 
     @property
     def full_table_name(self) -> FullyQualifiedName:

tests/packages/test_target_sqlite.py

@@ -13,7 +13,6 @@
 
 import pytest
 import sqlalchemy
-import sqlalchemy.exc
 from tap_hostile import TapHostile
 from tap_sqlite import SQLiteTap
 from target_sqlite import SQLiteSink, SQLiteTarget
@@ -206,11 +205,15 @@ def test_sync_sqlite_to_sqlite(
     assert line_num > 0, "No lines read."
 
 
-def test_sqlite_schema_addition(sqlite_sample_target: SQLTarget):
-    """Test that SQL-based targets attempt to create new schema.
+def test_sqlite_schema_addition(
+    sqlite_sample_target: SQLTarget,
+    sqlite_target_test_config: dict,
+):
+    """Test that a schema-qualified stream name loads successfully.
 
-    It should attempt to create a schema if one is included in stream name,
-    e.g. "schema_name-table_name".
+    SQLite has no `CREATE SCHEMA`, so a stream named "schema_name-table_name"
+    is folded by the target into a single physical table
+    "schema_name__table_name" rather than a real schema + table pair.
     """
     schema_name = f"test_schema_{str(uuid4()).split('-')[-1]}"
     table_name = f"zzz_tmp_{str(uuid4()).split('-')[-1]}"
@@ -234,16 +237,17 @@ def test_sqlite_schema_addition(sqlite_sample_target: SQLTarget):
             },
         ]
     )
-    # sqlite doesn't support schema creation
-    with pytest.raises(sqlalchemy.exc.OperationalError) as excinfo:
-        target_sync_test(
-            sqlite_sample_target,
-            input=StringIO(tap_output),
-            finalize=True,
-        )
-    # check the target at least tried to create the schema
-    assert isinstance(excinfo.value, sqlalchemy.exc.OperationalError)
-    assert excinfo.value.statement == f"CREATE SCHEMA {schema_name}"
+    target_sync_test(
+        sqlite_sample_target,
+        input=StringIO(tap_output),
+        finalize=True,
+    )
+    table = get_table(sqlite_target_test_config, f"{schema_name}__{table_name}")
+    with sqlalchemy.create_engine(
+        f"sqlite:///{sqlite_target_test_config['path_to_db']}",
+    ).connect() as conn:
+        rows = conn.execute(table.select()).fetchall()
+    assert [row.col_a for row in rows] == ["samplerow1"]
 
 
 def test_sqlite_column_addition(sqlite_sample_target: SQLTarget):

tests/sql/test_connector.py

@@ -537,6 +537,33 @@ def test_create_schema(self, connector: DummySQLConnector):
             mock_connect.return_value.__enter__.return_value.begin.assert_called_once()
             mock_connect.return_value.__enter__.return_value.execute.assert_called_once()
 
+    def test_schema_exists_for_real_schema(self, connector: DummySQLConnector):
+        """Test `schema_exists` returns True for existing schemas."""
+        # SQLite's built-in "main" schema always exists
+        assert connector.schema_exists("main") is True
+
+    def test_schema_exists_for_missing_schema(self, connector: DummySQLConnector):
+        """Test `schema_exists` returns False for missing schemas."""
+        assert connector.schema_exists("does_not_exist") is False
+
+    def test_prepare_schema_skips_creation_when_schema_exists(
+        self,
+        connector: DummySQLConnector,
+    ):
+        """Test `prepare_schema` does not create a schema that already exists."""
+        with mock.patch.object(connector, "create_schema") as mock_create_schema:
+            connector.prepare_schema("main")
+
+        mock_create_schema.assert_not_called()
+
+    def test_prepare_schema_creates_missing_schema(
+        self,
+        connector: DummySQLConnector,
+    ):
+        """Test `prepare_schema` attempts to create a schema that doesn't exist."""
+        with pytest.raises(sqlalchemy.exc.OperationalError, match="CREATE SCHEMA"):
+            connector.prepare_schema("does_not_exist")
+
     def test_column_rename(self, connector: DummySQLConnector):
         engine = connector._engine
         meta = sqlalchemy.MetaData()

tests/sql/test_sink.py

@@ -84,6 +84,25 @@ def sink(self, target: DummySQLTarget, schema: dict) -> DummySQLSink:
             key_properties=["id"],
         )
 
+    def test_setup_prepares_schema_when_present(
+        self,
+        target: DummySQLTarget,
+        schema: dict,
+    ):
+        """Test `Sink.setup()` calls `prepare_schema` when a schema name is present."""
+        sink = DummySQLSink(
+            target,
+            stream_name="main-foo",
+            schema=schema,
+            key_properties=["id"],
+        )
+
+        assert sink.schema_name == "main"
+
+        sink.setup()
+
+        assert sink.connector.table_exists(sink.full_table_name)
+
     def test_generate_insert_statement(self, sink: DummySQLSink, schema: dict):
         """Test that the insert statement is generated correctly."""
         stmt = sink.generate_insert_statement("foo", schema=schema)