SK-2954: Unify shared insert/validation/logging logic into common, add base insert response

Consolidates duplicated logic between v2 (PDB) and flowvault per architecture
review: shared validation (vault config, credentials, log level), LogLevel/Logger,
and insert field/table validation now live in common with per-variant message
injection; adds BaseInsertResponse alongside BaseInsertRequest so each variant's
InsertRequest/InsertResponse can extend a common base while keeping its own shape.
Also fixes flowvault's insert() response shape (drop redundant 'data'/'table',
flatten tokens, errors=None when empty) and a stale SDK_VERSION drift bug.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: saileshwar-skyflow

common/client/__init__.py

@@ -0,0 +1,189 @@
+import unittest
+
+from common.errors import SkyflowError
+from common.utils import LogLevel, SkyflowMessages
+from common.utils.logger import Logger
+from common.client.base_skyflow import make_skyflow_class
+
+
+class FakeVaultClient:
+    def __init__(self, config):
+        self._config = dict(config)
+        self.credentials = None
+        self.logger = None
+
+    def get_config(self):
+        return self._config
+
+    def update_config(self, config):
+        self._config.update(config)
+
+    def set_logger(self, log_level, logger):
+        self.logger = logger
+
+    def set_common_skyflow_credentials(self, credentials):
+        self.credentials = credentials
+
+
+class FakeVaultController:
+    def __init__(self, vault_client):
+        self.vault_client = vault_client
+
+
+class FakeConnection:
+    def __init__(self, vault_client):
+        self.vault_client = vault_client
+
+
+class FakeDetect:
+    def __init__(self, vault_client):
+        self.vault_client = vault_client
+
+
+def _noop_validate(logger, config):
+    return True
+
+
+def make_fake_skyflow(with_connections=False, with_detect=False):
+    kwargs = dict(
+        vault_client_cls=FakeVaultClient,
+        vault_controller_cls=FakeVaultController,
+        logger_cls=Logger,
+        default_log_level=LogLevel.ERROR,
+        skyflow_messages=SkyflowMessages,
+        validate_vault_config=_noop_validate,
+        validate_update_vault_config=_noop_validate,
+        validate_log_level=_noop_validate,
+        validate_credentials=_noop_validate,
+    )
+    if with_connections:
+        kwargs.update(
+            connection_cls=FakeConnection,
+            validate_connection_config=_noop_validate,
+            validate_update_connection_config=_noop_validate,
+        )
+    if with_detect:
+        kwargs['detect_cls'] = FakeDetect
+    return make_skyflow_class(**kwargs)
+
+
+VAULT_CONFIG = {"vault_id": "v1", "cluster_id": "c1", "credentials": {"token": "t"}}
+
+
+class TestMakeSkyflowClassBasics(unittest.TestCase):
+    def test_build_returns_instance_of_the_produced_class_not_the_template(self):
+        """Regression pin: build()/builder() must resolve to the specific class produced by
+        make_skyflow_class(), not the shared template -- two variants must never collide."""
+        SkyflowA = make_fake_skyflow()
+        SkyflowB = make_fake_skyflow()
+        client_a = SkyflowA.builder().add_vault_config(VAULT_CONFIG).build()
+        self.assertIsInstance(client_a, SkyflowA)
+        self.assertNotIsInstance(client_a, SkyflowB)
+
+    def test_two_produced_classes_do_not_share_hooks(self):
+        SkyflowWithDetect = make_fake_skyflow(with_detect=True)
+        SkyflowWithoutDetect = make_fake_skyflow(with_detect=False)
+        self.assertIsNotNone(SkyflowWithDetect.Builder._detect_cls)
+        self.assertIsNone(SkyflowWithoutDetect.Builder._detect_cls)
+
+    def test_vault_config_crud(self):
+        Skyflow = make_fake_skyflow()
+        builder = Skyflow.builder()
+        builder.add_vault_config(VAULT_CONFIG)
+        client = builder.build()
+
+        vault_config = client.get_vault_config("v1")
+        self.assertEqual(vault_config.get("vault_id"), "v1")
+
+        updated = dict(VAULT_CONFIG)
+        updated["cluster_id"] = "c2"
+        client.update_vault_config(updated)
+        self.assertEqual(client.get_vault_config("v1").get("cluster_id"), "c2")
+
+        client.remove_vault_config("v1")
+        with self.assertRaises(SkyflowError):
+            client.get_vault_config("v1")
+
+    def test_vault_returns_the_controller(self):
+        Skyflow = make_fake_skyflow()
+        client = Skyflow.builder().add_vault_config(VAULT_CONFIG).build()
+        self.assertIsInstance(client.vault("v1"), FakeVaultController)
+
+    def test_add_skyflow_credentials_and_update(self):
+        Skyflow = make_fake_skyflow()
+        client = Skyflow.builder().add_vault_config(VAULT_CONFIG).build()
+        client.add_skyflow_credentials({"token": "a"})
+        client.update_skyflow_credentials({"token": "b"})
+        # no assertion error means both delegate correctly to the same underlying builder path
+
+    def test_set_get_and_deprecated_update_log_level(self):
+        Skyflow = make_fake_skyflow()
+        client = Skyflow.builder().add_vault_config(VAULT_CONFIG).build()
+        client.set_log_level(LogLevel.INFO)
+        self.assertEqual(client.get_log_level(), LogLevel.INFO)
+
+        client.update_log_level(LogLevel.WARN)
+        self.assertEqual(client.get_log_level(), LogLevel.WARN)
+
+
+class TestConnectionAndDetectGating(unittest.TestCase):
+    def test_connection_methods_raise_when_connection_cls_not_supplied(self):
+        Skyflow = make_fake_skyflow()
+        client = Skyflow.builder().add_vault_config(VAULT_CONFIG).build()
+        with self.assertRaises(NotImplementedError):
+            client.connection()
+        with self.assertRaises(NotImplementedError):
+            client.add_connection_config({})
+        with self.assertRaises(NotImplementedError):
+            client.remove_connection_config("x")
+        with self.assertRaises(NotImplementedError):
+            client.update_connection_config({})
+        with self.assertRaises(NotImplementedError):
+            client.get_connection_config("x")
+
+    def test_detect_raises_when_detect_cls_not_supplied(self):
+        Skyflow = make_fake_skyflow()
+        client = Skyflow.builder().add_vault_config(VAULT_CONFIG).build()
+        with self.assertRaises(NotImplementedError):
+            client.detect()
+
+    def test_connection_config_crud_when_supplied(self):
+        Skyflow = make_fake_skyflow(with_connections=True)
+        connection_config = {"connection_id": "conn1", "connection_url": "https://x", "credentials": {"token": "t"}}
+        client = Skyflow.builder().add_connection_config(connection_config).build()
+
+        self.assertIsInstance(client.connection("conn1"), FakeConnection)
+
+        updated = dict(connection_config)
+        updated["connection_url"] = "https://y"
+        client.update_connection_config(updated)
+        self.assertEqual(client.get_connection_config("conn1").get("connection_url"), "https://y")
+
+        client.remove_connection_config("conn1")
+        with self.assertRaises(SkyflowError):
+            client.get_connection_config("conn1")
+
+    def test_detect_returns_detect_controller_when_supplied(self):
+        Skyflow = make_fake_skyflow(with_detect=True)
+        client = Skyflow.builder().add_vault_config(VAULT_CONFIG).build()
+        self.assertIsInstance(client.detect("v1"), FakeDetect)
+
+    def test_make_skyflow_class_requires_connection_validators_when_connection_cls_given(self):
+        with self.assertRaises(ValueError):
+            make_skyflow_class(
+                vault_client_cls=FakeVaultClient,
+                vault_controller_cls=FakeVaultController,
+                logger_cls=Logger,
+                default_log_level=LogLevel.ERROR,
+                skyflow_messages=SkyflowMessages,
+                validate_vault_config=_noop_validate,
+                validate_update_vault_config=_noop_validate,
+                validate_log_level=_noop_validate,
+                validate_credentials=_noop_validate,
+                connection_cls=FakeConnection,
+                # validate_connection_config/validate_update_connection_config omitted on purpose
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

common/client/base_skyflow.py

@@ -0,0 +1,169 @@
+import unittest
+
+from common.errors import SkyflowError
+from common.utils import SkyflowMessages, LogLevel, Env
+from common.utils.validations import (
+    validate_vault_config,
+    validate_update_vault_config,
+    validate_credentials,
+    validate_log_level,
+)
+
+VALID_VAULT_CONFIG = {
+    "vault_id": "vault123",
+    "cluster_id": "cluster1",
+    "env": Env.PROD,
+    "credentials": {"api_key": "sky-abcde-" + "f" * 32},
+}
+
+
+class FakeMessages:
+    """Stand-in message catalog to confirm validate_vault_config/etc. actually use the
+    `messages` param passed in, rather than silently falling back to common's own."""
+
+    class Error:
+        class _M:
+            def __init__(self, text):
+                self._text = text
+
+            @property
+            def value(self):
+                return self._text
+
+            def format(self, *args, **kwargs):
+                return self._text
+
+        EMPTY_VAULT_ID = _M("FAKE: empty vault id")
+        INVALID_VAULT_ID = _M("FAKE: invalid vault id")
+        EMPTY_CLUSTER_ID = _M("FAKE: empty cluster id")
+        INVALID_CLUSTER_ID = _M("FAKE: invalid cluster id")
+        EMPTY_CREDENTIALS = _M("FAKE: empty credentials")
+        INVALID_ENV = _M("FAKE: invalid env")
+        INVALID_KEY = _M("FAKE: invalid key")
+        INVALID_LOG_LEVEL = _M("FAKE: invalid log level")
+        INVALID_CREDENTIALS = _M("FAKE: invalid credentials")
+        INVALID_CREDENTIALS_IN_CONFIG = _M("FAKE: invalid credentials in config")
+
+    class ErrorLogs:
+        class _M:
+            def __init__(self, text):
+                self._text = text
+
+            @property
+            def value(self):
+                return self._text
+
+        VAULTID_IS_REQUIRED = _M("fake log")
+        CLUSTER_ID_IS_REQUIRED = _M("fake log")
+        CONNECTION_ID_IS_REQUIRED = _M("fake log")
+        INVALID_CONNECTION_URL = _M("fake log")
+        EMPTY_VAULTID = _M("fake log")
+        EMPTY_CLUSTER_ID = _M("fake log")
+        EMPTY_CONNECTION_ID = _M("fake log")
+        EMPTY_CONNECTION_URL = _M("fake log")
+        EMPTY_CREDENTIALS_PATH = _M("fake log")
+        EMPTY_CREDENTIALS_STRING = _M("fake log")
+        EMPTY_TOKEN_VALUE = _M("fake log")
+        EMPTY_API_KEY_VALUE = _M("fake log")
+        INVALID_KEY = _M("fake log")
+        INVALID_LOG_LEVEL = _M("fake log")
+        ENV_IS_REQUIRED = _M("fake log")
+
+    class Info:
+        class _M:
+            def __init__(self, text):
+                self._text = text
+
+            @property
+            def value(self):
+                return self._text
+
+        VALIDATING_VAULT_CONFIG = _M("fake info")
+
+
+class TestValidateVaultConfig(unittest.TestCase):
+    def test_valid_config_passes(self):
+        self.assertTrue(validate_vault_config(None, dict(VALID_VAULT_CONFIG)))
+
+    def test_missing_vault_id_raises(self):
+        config = dict(VALID_VAULT_CONFIG)
+        del config["vault_id"]
+        with self.assertRaises(SkyflowError):
+            validate_vault_config(None, config)
+
+    def test_unknown_key_raises(self):
+        config = dict(VALID_VAULT_CONFIG)
+        config["unexpected"] = True
+        with self.assertRaises(SkyflowError):
+            validate_vault_config(None, config)
+
+    def test_empty_credentials_raises(self):
+        config = dict(VALID_VAULT_CONFIG)
+        config["credentials"] = {}
+        with self.assertRaises(SkyflowError):
+            validate_vault_config(None, config)
+
+    def test_credentials_are_validated(self):
+        config = dict(VALID_VAULT_CONFIG)
+        config["credentials"] = {"api_key": "not-a-valid-key"}
+        with self.assertRaises(SkyflowError):
+            validate_vault_config(None, config)
+
+    def test_uses_injected_messages_for_raised_error(self):
+        config = dict(VALID_VAULT_CONFIG)
+        del config["vault_id"]
+        with self.assertRaises(SkyflowError) as ctx:
+            validate_vault_config(None, config, messages=FakeMessages)
+        self.assertIn("FAKE", ctx.exception.message)
+
+    def test_defaults_to_common_messages_when_not_injected(self):
+        config = dict(VALID_VAULT_CONFIG)
+        del config["vault_id"]
+        with self.assertRaises(SkyflowError) as ctx:
+            validate_vault_config(None, config)
+        self.assertEqual(ctx.exception.message, SkyflowMessages.Error.INVALID_VAULT_ID.value)
+
+
+class TestValidateUpdateVaultConfig(unittest.TestCase):
+    def test_valid_update_passes(self):
+        self.assertTrue(validate_update_vault_config(None, dict(VALID_VAULT_CONFIG)))
+
+    def test_credentials_required_on_update(self):
+        """Unlike validate_vault_config, credentials are mandatory here."""
+        config = dict(VALID_VAULT_CONFIG)
+        del config["credentials"]
+        with self.assertRaises(SkyflowError):
+            validate_update_vault_config(None, config)
+
+    def test_uses_injected_messages(self):
+        config = dict(VALID_VAULT_CONFIG)
+        del config["credentials"]
+        with self.assertRaises(SkyflowError) as ctx:
+            validate_update_vault_config(None, config, messages=FakeMessages)
+        self.assertIn("FAKE", ctx.exception.message)
+
+
+class TestValidateCredentialsMessageInjection(unittest.TestCase):
+    def test_uses_injected_messages(self):
+        with self.assertRaises(SkyflowError) as ctx:
+            validate_credentials(None, {}, messages=FakeMessages)
+        self.assertIn("FAKE", ctx.exception.message)
+
+    def test_defaults_to_common_messages(self):
+        with self.assertRaises(SkyflowError) as ctx:
+            validate_credentials(None, {})
+        self.assertEqual(ctx.exception.message, SkyflowMessages.Error.INVALID_CREDENTIALS.value)
+
+
+class TestValidateLogLevelMessageInjection(unittest.TestCase):
+    def test_valid_log_level_passes(self):
+        validate_log_level(None, LogLevel.INFO)  # should not raise
+
+    def test_uses_injected_messages(self):
+        with self.assertRaises(SkyflowError) as ctx:
+            validate_log_level(None, "not-a-log-level", messages=FakeMessages)
+        self.assertIn("FAKE", ctx.exception.message)
+
+
+if __name__ == "__main__":
+    unittest.main()