From ea727267b8047caa1e898ca44a4d3d57e16f45f1 Mon Sep 17 00:00:00 2001 From: Akash Malbari Date: Tue, 2 Jun 2026 09:41:33 -0400 Subject: [PATCH 1/4] fix(targets): reuse SQLAlchemy inspector reflection cache --- singer_sdk/sql/connector.py | 33 ++++++++-- tests/sql/test_connector.py | 119 ++++++++++++++++++++++++++++++++++++ 2 files changed, 148 insertions(+), 4 deletions(-) diff --git a/singer_sdk/sql/connector.py b/singer_sdk/sql/connector.py index 5e83ebe87e..71498f71b6 100644 --- a/singer_sdk/sql/connector.py +++ b/singer_sdk/sql/connector.py @@ -621,6 +621,7 @@ def __init__( self._config = config or {} self._sqlalchemy_url: str | None = sqlalchemy_url or None self._tables_prepared: dict[str, bool] = {} + self._cached_inspector: reflection.Inspector | None = None @property def config(self) -> Mapping[str, t.Any]: @@ -894,6 +895,25 @@ def _engine(self) -> sa.Engine: self._cached_engine = self.create_engine() return self._cached_engine + @property + def _inspector(self) -> reflection.Inspector: + """Return the cached SQLAlchemy inspector for this connector.""" + if self._cached_inspector is None: + self._cached_inspector = sa.inspect(self._engine) + return self._cached_inspector + + def clear_reflection_cache(self) -> None: + """Clear cached SQLAlchemy reflection state, if an inspector exists.""" + if self._cached_inspector is None: + return + + clear_cache = getattr(self._cached_inspector, "clear_cache", None) + if clear_cache: + clear_cache() + return + + self._cached_inspector.info_cache.clear() + def create_engine(self) -> sa.Engine: """Creates and returns a new engine. Do not call outside of _engine. @@ -1206,7 +1226,7 @@ def table_exists(self, full_table_name: str | FullyQualifiedName) -> bool: """ _, schema_name, table_name = self.parse_full_table_name(full_table_name) - return sa.inspect(self._engine).has_table(table_name, schema_name) + return self._inspector.has_table(table_name, schema_name) def schema_exists(self, schema_name: str) -> bool: """Determine if the target database schema already exists. @@ -1217,7 +1237,7 @@ def schema_exists(self, schema_name: str) -> bool: Returns: True if the database schema exists, False if not. """ - schemas = set(sa.inspect(self._engine).get_schema_names()) + schemas = set(self._inspector.get_schema_names()) return schema_name in schemas def get_table_columns( @@ -1235,8 +1255,7 @@ def get_table_columns( An ordered list of column objects. """ _, schema_name, table_name = self.parse_full_table_name(full_table_name) - inspector = sa.inspect(self._engine) - columns = inspector.get_columns(table_name, schema_name) + columns = self._inspector.get_columns(table_name, schema_name) columns_dict: dict[str, sa.Column] = { col_meta["name"]: sa.Column( @@ -1300,6 +1319,7 @@ def create_schema(self, schema_name: str) -> None: """ with self._connect() as conn, conn.begin(): conn.execute(ddl.CreateSchema(schema_name)) + self.clear_reflection_cache() def create_empty_table( self, @@ -1354,6 +1374,7 @@ def create_empty_table( _ = sa.Table(table_name, meta, *columns, *table_args) meta.create_all(self._engine) + self.clear_reflection_cache() def _create_empty_column( self, @@ -1382,6 +1403,7 @@ def _create_empty_column( ) with self._connect() as conn, conn.begin(): conn.execute(column_add_ddl) + self.clear_reflection_cache() def prepare_schema(self, schema_name: str) -> None: """Create the target database schema. @@ -1426,6 +1448,7 @@ def prepare_table( meta = sa.MetaData() table = sa.Table(table_name, meta, schema=schema_name) table.drop(self._engine, checkfirst=True) + self.clear_reflection_cache() self.logger.info("Creating empty table %s", full_table_name) self.create_empty_table( full_table_name=full_table_name, @@ -1521,6 +1544,7 @@ def rename_column( ) with self._connect() as conn, conn.begin(): conn.execute(column_rename_ddl) + self.clear_reflection_cache() def merge_sql_types( self, @@ -1828,6 +1852,7 @@ def _adapt_column_type( ) with self._connect() as conn, conn.begin(): conn.execute(alter_column_ddl) + self.clear_reflection_cache() def serialize_json(self, obj: object) -> str: # noqa: PLR6301 """Serialize an object to a JSON string. diff --git a/tests/sql/test_connector.py b/tests/sql/test_connector.py index 2cc8821992..edb83fe2d5 100644 --- a/tests/sql/test_connector.py +++ b/tests/sql/test_connector.py @@ -198,6 +198,60 @@ def test_engine_creates_and_returns_cached_engine(self, connector): engine2 = connector._cached_engine assert engine1 is engine2 + def test_reflection_methods_reuse_cached_inspector( + self, + connector: SQLConnector, + ): + engine = connector._engine + inspector = mock.Mock() + inspector.has_table.return_value = True + inspector.get_schema_names.return_value = ["main"] + inspector.get_columns.return_value = [ + {"name": "id", "type": sqlalchemy.Integer(), "nullable": False}, + ] + + with mock.patch( + "singer_sdk.sql.connector.sa.inspect", + return_value=inspector, + ) as mock_inspect: + assert connector.table_exists("main.test_table") + assert connector.schema_exists("main") + columns = connector.get_table_columns("main.test_table") + assert connector.table_exists("main.test_table") + + mock_inspect.assert_called_once_with(engine) + assert connector._cached_inspector is inspector + assert list(columns) == ["id"] + inspector.has_table.assert_has_calls( + [ + mock.call("test_table", "main"), + mock.call("test_table", "main"), + ], + ) + inspector.get_schema_names.assert_called_once_with() + inspector.get_columns.assert_called_once_with("test_table", "main") + + def test_clear_reflection_cache_does_not_create_inspector( + self, + connector: SQLConnector, + ): + with mock.patch("singer_sdk.sql.connector.sa.inspect") as mock_inspect: + connector.clear_reflection_cache() + + mock_inspect.assert_not_called() + + def test_clear_reflection_cache_clears_cached_inspector( + self, + connector: SQLConnector, + ): + inspector = connector._inspector + inspector.info_cache["sentinel"] = object() + + connector.clear_reflection_cache() + + assert inspector.info_cache == {} + assert connector._cached_inspector is inspector + def test_deprecated_functions_warn(self, connector: SQLConnector): with pytest.deprecated_call(): connector.create_sqlalchemy_engine() @@ -374,6 +428,34 @@ def test_create_empty_table_primary_key_order( with connector._engine.connect() as conn, conn.begin(): conn.execute(sqlalchemy.text(f"DROP TABLE {table_name}")) + def test_create_empty_table_clears_cached_negative_reflection( + self, + connector: SQLConnector, + ): + table_name = "test_reflection_cache" + schema = { + "type": "object", + "properties": { + "id": {"type": "integer"}, + }, + } + + assert not connector.table_exists(table_name) + inspector = connector._cached_inspector + assert inspector is not None + + connector.create_empty_table( + full_table_name=table_name, + schema=schema, + primary_keys=[], + ) + + assert connector._cached_inspector is inspector + assert connector.table_exists(table_name) + + with connector._engine.connect() as conn, conn.begin(): + conn.execute(sqlalchemy.text(f"DROP TABLE {table_name}")) + def test_create_empty_table_partial_primary_keys(self, connector: SQLConnector): """Test that only existing columns are included in primary key constraint.""" schema = { @@ -529,6 +611,10 @@ def test_create_schema(self, connector: DummySQLConnector): # CREATE SCHEMA is not supported by SQLite, so we mock with ( mock.patch.object(connector, "_connect") as mock_connect, + mock.patch.object( + connector, + "clear_reflection_cache", + ) as mock_clear_reflection_cache, mock.patch.object(mock_connect, "begin"), mock.patch.object(mock_connect, "execute"), ): @@ -536,6 +622,7 @@ def test_create_schema(self, connector: DummySQLConnector): mock_connect.assert_called_once() mock_connect.return_value.__enter__.return_value.begin.assert_called_once() mock_connect.return_value.__enter__.return_value.execute.assert_called_once() + mock_clear_reflection_cache.assert_called_once_with() def test_column_rename(self, connector: DummySQLConnector): engine = connector._engine @@ -548,12 +635,39 @@ def test_column_rename(self, connector: DummySQLConnector): ) meta.create_all(engine) + assert list(connector.get_table_columns("test_table")) == ["id", "old_name"] + connector.rename_column("test_table", "old_name", "new_name") + assert list(connector.get_table_columns("test_table")) == ["id", "new_name"] + with engine.connect() as conn: result = conn.execute(sqlalchemy.text("SELECT * FROM test_table")) assert result.keys() == ["id", "new_name"] + def test_create_empty_column_clears_reflection_cache( + self, + connector: DummySQLConnector, + ): + engine = connector._engine + meta = sqlalchemy.MetaData() + _ = sqlalchemy.Table( + "test_table", + meta, + sqlalchemy.Column("id", sqlalchemy.Integer), + ) + meta.create_all(engine) + + assert list(connector.get_table_columns("test_table")) == ["id"] + + connector._create_empty_column( + "test_table", + "name", + sqlalchemy.String(), + ) + + assert list(connector.get_table_columns("test_table")) == ["id", "name"] + def test_adapt_column_type(self, connector: DummySQLConnector): engine = connector._engine meta = sqlalchemy.MetaData() @@ -568,6 +682,10 @@ def test_adapt_column_type(self, connector: DummySQLConnector): # Changing the column type is not supported by SQLite, so we mock with ( mock.patch.object(connector, "_connect") as mock_connect, + mock.patch.object( + connector, + "clear_reflection_cache", + ) as mock_clear_reflection_cache, mock.patch.object(mock_connect, "begin"), mock.patch.object(mock_connect, "execute"), ): @@ -582,6 +700,7 @@ def test_adapt_column_type(self, connector: DummySQLConnector): str(ddl.compile()) == "ALTER TABLE test_table ALTER COLUMN name TYPE VARCHAR" ) + mock_clear_reflection_cache.assert_called_once_with() @pytest.mark.parametrize( "exclude_schemas,expected_streams", From 2072c9197193cb1362bb50fcaaa83975bb95f544 Mon Sep 17 00:00:00 2001 From: Akash Malbari Date: Wed, 3 Jun 2026 08:55:41 -0400 Subject: [PATCH 2/4] Fix SQLAlchemy inspector reflection cache reuse --- AGENTS.md | 124 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 124 insertions(+) create mode 100644 AGENTS.md diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000000..edeb323fc3 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,124 @@ +# AGENTS.md + +This is the Meltano Singer SDK, a Python framework for building Singer taps (data extractors) and targets (data loaders). + +## Development Setup + +```bash +uv sync --all-groups --all-extras --all-packages # Full development environment +``` + +## Common Commands + +### Testing + +```bash +nox -s tests # Run core tests +nox -s test-contrib # Run contrib (experimental) tests +nox -s test-packages # Run package integration tests +nox -t typing # Type checking with mypy +``` + +### Linting & Formatting + +```bash +pre-commit run --all # Run all pre-commit hooks +ruff check --fix # Lint and auto-fix +ruff format # Format code +``` + +### Documentation + +```bash +nox -s docs # Build Sphinx documentation +``` + +## Code Style + +- Required: `from __future__ import annotations` at top of every file +- Use `typing as t` for type imports (abbreviated import per ruff config) +- Google-style docstrings with full parameter documentation +- Line length: 88 characters +- Ruff handles all formatting and linting + +## Exceptions + +All SDK exceptions live in `singer_sdk/exceptions.py` and inherit from `SingerSDKError`. +Choose the right base class by recovery strategy: + +| Situation | Raise | +|---|---| +| HTTP/API error, must abort sync | `FatalAPIError` | +| HTTP/API error, safe to retry | `RetriableAPIError` | +| HTTP/API error, expected / skip silently | `IgnorableAPIError` | +| Config value is wrong | `ConfigValidationError` | +| Discovery / catalog problem | `DiscoveryError` (or a subclass) | +| Stream map config or expression fails | `MappingError` (or a subclass) | + +When adding a new exception: + +1. Place it in `singer_sdk/exceptions.py` — never define public exceptions in other files. +1. Inherit from the appropriate intermediate base (`FatalSyncError`, `RetriableSyncError`, + `IgnorableSyncError`, `DataError`, `ConfigurationError`, `MappingError`, etc.) rather + than from `Exception` or `SingerSDKError` directly. +1. Add it to `__all__` in that file. +1. Add `issubclass` assertions to `tests/core/test_exceptions.py`. + +See `docs/implementation/errors/hierarchy.md` for the full hierarchy and +`docs/implementation/errors/design.md` for the design rationale. + +## Deprecation Warnings + +Three warning classes in `singer_sdk/helpers/_compat.py`: + +| Class | Base | When to use | +|---|---|---| +| `SingerSDKDeprecationWarning` | `DeprecationWarning` | Removal version is known | +| `SingerSDKPendingDeprecationWarning` | `PendingDeprecationWarning` | No committed removal timeline; silenced by default | +| `SingerSDKPythonEOLWarning` | `FutureWarning` | Python version nearing/past EOL | + +Use `deprecated(msg, *, category=SingerSDKDeprecationWarning, stacklevel=1)` as a decorator factory for deprecating classes and functions. `category` is **required** — it is appended to the message automatically. Never embed the version in the message string manually. + +```python +@deprecated( + "Use Bar instead. Foo will be removed in .", + category=SingerSDKDeprecationWarning, +) +class Foo: ... +``` + +For `warnings.warn` call sites with no committed removal timeline, use `SingerSDKPendingDeprecationWarning` directly: + +```python +warnings.warn("...", SingerSDKPendingDeprecationWarning, stacklevel=2) +``` + +**Deprecation policy**: at least 3 months / 3 feature releases notice, named removal version required. See `docs/release_process.md` and `docs/deprecation.md`. + +## Architecture + +The SDK uses abstract base classes for its plugin system: + +- `Tap` / `Target` - Main plugin entry points +- `Stream` / `RESTStream` / `GraphQLStream` / `SQLStream` - Data extraction +- `Sink` / `BatchSink` / `SQLSink` - Data loading +- `SQLConnector` - Database connections + +## Testing + +- pytest-based with custom pytest plugin (`singer_testing`) +- Test markers: `contrib`, `external`, `packages`, `snapshot` +- Platform markers: `@pytest.mark.darwin`, `@pytest.mark.linux`, `@pytest.mark.windows` +- Standard test suites: `singer_sdk.testing.get_standard_tap_tests()` / `get_standard_target_tests()` + +## Project Structure + +- `singer_sdk/` - Main SDK package +- `tests/` - Test suite (core, contrib, external, packages) +- `packages/` - Reference implementations for E2E testing +- `cookiecutter/` - Templates for scaffolding new taps/targets +- `docs/` - Sphinx documentation + +## Python Version + +Supports Python 3.10-3.14. Primary development version is 3.14. From bc42216d0b95ca32e257404fcc879c719f390c99 Mon Sep 17 00:00:00 2001 From: Akash Malbari Date: Wed, 3 Jun 2026 10:29:03 -0400 Subject: [PATCH 3/4] Cover clear_reflection_cache fallback path --- tests/sql/test_connector.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/tests/sql/test_connector.py b/tests/sql/test_connector.py index edb83fe2d5..bef2f22e3e 100644 --- a/tests/sql/test_connector.py +++ b/tests/sql/test_connector.py @@ -252,6 +252,21 @@ def test_clear_reflection_cache_clears_cached_inspector( assert inspector.info_cache == {} assert connector._cached_inspector is inspector + def test_clear_reflection_cache_falls_back_to_info_cache( + self, + connector: SQLConnector, + ): + class FakeInspector: + def __init__(self) -> None: + self.info_cache = {"tables": "cached"} + + fake_inspector = FakeInspector() + connector._cached_inspector = fake_inspector + + connector.clear_reflection_cache() + + assert fake_inspector.info_cache == {} + def test_deprecated_functions_warn(self, connector: SQLConnector): with pytest.deprecated_call(): connector.create_sqlalchemy_engine() From 2c9643a0596669029a510aaad46ff61efdd5cc3c Mon Sep 17 00:00:00 2001 From: Akash Malbari Date: Fri, 26 Jun 2026 09:36:52 -0400 Subject: [PATCH 4/4] Auto-clear SQL reflection cache after DDL --- singer_sdk/sql/connector.py | 28 ++++++++++++++++------ tests/sql/test_connector.py | 47 +++++++++++++++++++++++++++++-------- 2 files changed, 58 insertions(+), 17 deletions(-) diff --git a/singer_sdk/sql/connector.py b/singer_sdk/sql/connector.py index 18e3bb7d07..151e146bce 100644 --- a/singer_sdk/sql/connector.py +++ b/singer_sdk/sql/connector.py @@ -13,6 +13,7 @@ import sqlalchemy as sa import sqlalchemy.types +from sqlalchemy import event from sqlalchemy.engine import reflection from sqlalchemy.sql import ddl @@ -893,11 +894,16 @@ def _engine(self) -> sa.Engine: """ if not self._cached_engine: self._cached_engine = self.create_engine() + event.listen( + self._cached_engine, + "after_cursor_execute", + self._clear_reflection_cache_after_ddl, + ) return self._cached_engine @property def _inspector(self) -> reflection.Inspector: - """Return the cached SQLAlchemy inspector for this connector.""" + """Cached SQLAlchemy inspector for this connector.""" if self._cached_inspector is None: self._cached_inspector = sa.inspect(self._engine) return self._cached_inspector @@ -914,6 +920,20 @@ def clear_reflection_cache(self) -> None: self._cached_inspector.info_cache.clear() + def _clear_reflection_cache_after_ddl( + self, + _conn: sa.Connection, + _cursor: object, + statement: str, + _parameters: object, + _context: object, + _executemany: bool, + ) -> None: + """Clear cached reflection state after SQLAlchemy DDL executes.""" + statement_prefix = statement.lstrip().partition(" ")[0].upper() + if statement_prefix in {"ALTER", "CREATE", "DROP"}: + self.clear_reflection_cache() + def create_engine(self) -> sa.Engine: """Creates and returns a new engine. Do not call outside of _engine. @@ -1319,7 +1339,6 @@ def create_schema(self, schema_name: str) -> None: """ with self._connect() as conn, conn.begin(): conn.execute(ddl.CreateSchema(schema_name)) - self.clear_reflection_cache() def create_empty_table( self, @@ -1374,7 +1393,6 @@ def create_empty_table( _ = sa.Table(table_name, meta, *columns, *table_args) meta.create_all(self._engine) - self.clear_reflection_cache() def _create_empty_column( self, @@ -1403,7 +1421,6 @@ def _create_empty_column( ) with self._connect() as conn, conn.begin(): conn.execute(column_add_ddl) - self.clear_reflection_cache() def prepare_schema(self, schema_name: str) -> None: """Create the target database schema. @@ -1448,7 +1465,6 @@ def prepare_table( meta = sa.MetaData() table = sa.Table(table_name, meta, schema=schema_name) table.drop(self._engine, checkfirst=True) - self.clear_reflection_cache() self.logger.info("Creating empty table %s", full_table_name) self.create_empty_table( full_table_name=full_table_name, @@ -1544,7 +1560,6 @@ def rename_column( ) with self._connect() as conn, conn.begin(): conn.execute(column_rename_ddl) - self.clear_reflection_cache() def merge_sql_types( self, @@ -1852,7 +1867,6 @@ def _adapt_column_type( ) with self._connect() as conn, conn.begin(): conn.execute(alter_column_ddl) - self.clear_reflection_cache() def serialize_json(self, obj: object) -> str: # noqa: PLR6301 """Serialize an object to a JSON string. diff --git a/tests/sql/test_connector.py b/tests/sql/test_connector.py index bef2f22e3e..8fe9fccca6 100644 --- a/tests/sql/test_connector.py +++ b/tests/sql/test_connector.py @@ -626,10 +626,6 @@ def test_create_schema(self, connector: DummySQLConnector): # CREATE SCHEMA is not supported by SQLite, so we mock with ( mock.patch.object(connector, "_connect") as mock_connect, - mock.patch.object( - connector, - "clear_reflection_cache", - ) as mock_clear_reflection_cache, mock.patch.object(mock_connect, "begin"), mock.patch.object(mock_connect, "execute"), ): @@ -637,7 +633,6 @@ def test_create_schema(self, connector: DummySQLConnector): mock_connect.assert_called_once() mock_connect.return_value.__enter__.return_value.begin.assert_called_once() mock_connect.return_value.__enter__.return_value.execute.assert_called_once() - mock_clear_reflection_cache.assert_called_once_with() def test_column_rename(self, connector: DummySQLConnector): engine = connector._engine @@ -683,6 +678,43 @@ def test_create_empty_column_clears_reflection_cache( assert list(connector.get_table_columns("test_table")) == ["id", "name"] + def test_sqlalchemy_ddl_clears_reflection_cache_for_overrides(self): + class CustomConnector(DummySQLConnector): + @override + def _create_empty_column( + self, + full_table_name: str | FullyQualifiedName, + column_name: str, + sql_type: sqlalchemy.types.TypeEngine, + ) -> None: + column_add_ddl = self.get_column_add_ddl( + table_name=full_table_name, + column_name=column_name, + column_type=sql_type, + ) + with self._connect() as conn, conn.begin(): + conn.execute(column_add_ddl) + + connector = CustomConnector(config={"sqlalchemy_url": "sqlite:///"}) + engine = connector._engine + meta = sqlalchemy.MetaData() + _ = sqlalchemy.Table( + "test_table", + meta, + sqlalchemy.Column("id", sqlalchemy.Integer), + ) + meta.create_all(engine) + + assert list(connector.get_table_columns("test_table")) == ["id"] + + connector._create_empty_column( + "test_table", + "name", + sqlalchemy.String(), + ) + + assert list(connector.get_table_columns("test_table")) == ["id", "name"] + def test_adapt_column_type(self, connector: DummySQLConnector): engine = connector._engine meta = sqlalchemy.MetaData() @@ -697,10 +729,6 @@ def test_adapt_column_type(self, connector: DummySQLConnector): # Changing the column type is not supported by SQLite, so we mock with ( mock.patch.object(connector, "_connect") as mock_connect, - mock.patch.object( - connector, - "clear_reflection_cache", - ) as mock_clear_reflection_cache, mock.patch.object(mock_connect, "begin"), mock.patch.object(mock_connect, "execute"), ): @@ -715,7 +743,6 @@ def test_adapt_column_type(self, connector: DummySQLConnector): str(ddl.compile()) == "ALTER TABLE test_table ALTER COLUMN name TYPE VARCHAR" ) - mock_clear_reflection_cache.assert_called_once_with() @pytest.mark.parametrize( "exclude_schemas,expected_streams",