diff --git a/packages/meltano-target-sqlite/target_sqlite/target.py b/packages/meltano-target-sqlite/target_sqlite/target.py index 7c8ded26e..84ef96ddc 100644 --- a/packages/meltano-target-sqlite/target_sqlite/target.py +++ b/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 `--` or `-
` 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.""" diff --git a/singer_sdk/sql/sink.py b/singer_sdk/sql/sink.py index 2c0a3d282..de03ec785 100644 --- a/singer_sdk/sql/sink.py +++ b/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: `
`, `-
`, + or `--
`. + + 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 + `--
` 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 `--
` 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 `--
` 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: diff --git a/tests/packages/test_target_sqlite.py b/tests/packages/test_target_sqlite.py index 8c88ba137..52fe193bc 100644 --- a/tests/packages/test_target_sqlite.py +++ b/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): diff --git a/tests/sql/test_connector.py b/tests/sql/test_connector.py index 9fe4b80f8..d81558e6a 100644 --- a/tests/sql/test_connector.py +++ b/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() diff --git a/tests/sql/test_sink.py b/tests/sql/test_sink.py index f4feac02b..9f81a5fd1 100644 --- a/tests/sql/test_sink.py +++ b/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)