feat: add check_config_during_discover flag for targeted config validation control

Co-Authored-By: Aaron <AJ> Steers <[email protected]>

Author: devin-ai-integration[bot]

CHANGELOG.md

@@ -4,6 +4,10 @@ Newer updates can be found here: [GitHub Release Notes](https://github.com/airby
 
 # Changelog
 
+## Unreleased
+
+- Added `check_config_during_discover` flag to declarative sources to skip config validation during discovery for sources with DynamicSchemaLoader
+
 ## 6.5.2
 
 bugfix: Ensure that streams with partition router are not executed concurrently

airbyte_cdk/entrypoint.py

@@ -149,7 +149,8 @@ def run(self, parsed_args: argparse.Namespace) -> Iterable[str]:
                 elif (
                     cmd == "discover"
                     and not parsed_args.config
-                    and not self.source.check_config_against_spec
+                    and hasattr(self.source, "check_config_during_discover")
+                    and self.source.check_config_during_discover
                 ):
                     # Connector supports unprivileged discover
                     empty_config: dict[str, Any] = {}
@@ -240,7 +241,7 @@ def discover(
         self, source_spec: ConnectorSpecification, config: TConfig
     ) -> Iterable[AirbyteMessage]:
         self.set_up_secret_filter(config, source_spec.connectionSpecification)
-        if self.source.check_config_against_spec:
+        if not hasattr(self.source, "check_config_during_discover") or not self.source.check_config_during_discover:
             self.validate_connection(source_spec, config)
         catalog = self.source.discover(self.logger, config)
 

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -109,7 +109,8 @@ def __init__(
         self._config = config or {}
         self._validate_source()
 
-        self.check_config_against_spec = not self._uses_dynamic_schema_loader()
+        self.check_config_during_discover = self._uses_dynamic_schema_loader()
+        self.check_config_against_spec = True
 
     @property
     def resolved_manifest(self) -> Mapping[str, Any]:

unit_tests/sources/declarative/test_manifest_declarative_source_dynamic_schema.py

@@ -10,8 +10,8 @@
 from airbyte_cdk.sources.utils.schema_helpers import check_config_against_spec_or_exit
 
 
-def test_check_config_against_spec_with_dynamic_schema_loader():
-    """Test that check_config_against_spec is False when DynamicSchemaLoader is used."""
+def test_check_config_during_discover_with_dynamic_schema_loader():
+    """Test that check_config_during_discover is True when DynamicSchemaLoader is used."""
     source_config = {
         "type": "DeclarativeSource",
         "check": {"type": "CheckStream"},
@@ -41,11 +41,12 @@ def test_check_config_against_spec_with_dynamic_schema_loader():
 
     source = ManifestDeclarativeSource(source_config=source_config)
 
-    assert source.check_config_against_spec is False
+    assert source.check_config_during_discover is True
+    assert source.check_config_against_spec is True
 
 
-def test_check_config_against_spec_without_dynamic_schema_loader():
-    """Test that check_config_against_spec is True when DynamicSchemaLoader is not used."""
+def test_check_config_during_discover_without_dynamic_schema_loader():
+    """Test that check_config_during_discover is False when DynamicSchemaLoader is not used."""
     source_config = {
         "type": "DeclarativeSource",
         "check": {"type": "CheckStream"},
@@ -64,6 +65,9 @@ def test_check_config_against_spec_without_dynamic_schema_loader():
     }
 
     source = ManifestDeclarativeSource(source_config=source_config)
+    
+    assert source.check_config_during_discover is False
+    assert source.check_config_against_spec is True
 
 
 @patch(
@@ -109,7 +113,8 @@ def test_discover_with_dynamic_schema_loader_no_config(mock_streams):
 
     source = ManifestDeclarativeSource(source_config=source_config)
 
-    assert source.check_config_against_spec is False
+    assert source.check_config_during_discover is True
+    assert source.check_config_against_spec is True
 
     logger = MagicMock()
     catalog = source.discover(logger, {})
@@ -152,6 +157,7 @@ def test_discover_without_dynamic_schema_loader_no_config(mock_streams):
 
     source = ManifestDeclarativeSource(source_config=source_config)
 
+    assert source.check_config_during_discover is False
     assert source.check_config_against_spec is True
 
     logger = MagicMock()
@@ -161,4 +167,5 @@ def test_discover_without_dynamic_schema_loader_no_config(mock_streams):
     assert len(catalog.streams) == 1
     assert catalog.streams[0].name == "test_static_stream"
 
+    assert source.check_config_during_discover is False
     assert source.check_config_against_spec is True