perf(declarative): skip runtime metaschema check of the component schema

airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py

@@ -12,8 +12,8 @@
 
 import orjson
 import yaml
-from jsonschema.exceptions import ValidationError
-from jsonschema.validators import validate
+from jsonschema.exceptions import ValidationError, best_match
+from jsonschema.validators import validator_for
 from packaging.version import InvalidVersion, Version
 
 from airbyte_cdk.config_observation import create_connector_config_control_message
@@ -447,12 +447,23 @@ def _validate_source(self) -> None:
         Validates the connector manifest against the declarative component schema
         """
 
+        # The component schema ships with the CDK and is checked against its metaschema in unit
+        # tests, so a valid manifest skips `check_schema`. On failure the schema is checked first,
+        # so a broken schema is reported as `SchemaError` rather than blamed on the manifest.
+        # `best_match` keeps the error `jsonschema.validate` would raise.
+        validator_class = validator_for(self._declarative_component_schema)
         try:
-            validate(self._source_config, self._declarative_component_schema)
-        except ValidationError as e:
+            error = best_match(
+                validator_class(self._declarative_component_schema).iter_errors(self._source_config)
+            )
+        except Exception:
+            validator_class.check_schema(self._declarative_component_schema)
+            raise
+        if error is not None:
+            validator_class.check_schema(self._declarative_component_schema)
             raise ValidationError(
                 "Validation against json schema defined in declarative_component_schema.yaml schema failed"
-            ) from e
+            ) from error
 
         cdk_version_str = metadata.version("airbyte_cdk")
         cdk_version = self._parse_version(cdk_version_str, "airbyte-cdk")

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -22,8 +22,8 @@
 import orjson
 import yaml
 from airbyte_protocol_dataclasses.models import AirbyteStreamStatus, Level, StreamDescriptor
-from jsonschema.exceptions import ValidationError
-from jsonschema.validators import validate
+from jsonschema.exceptions import ValidationError, best_match
+from jsonschema.validators import validator_for
 
 from airbyte_cdk.config_observation import create_connector_config_control_message
 from airbyte_cdk.connector_builder.models import (
@@ -345,12 +345,23 @@ def _validate_source(self) -> None:
         Validates the connector manifest against the declarative component schema
         """
 
+        # The component schema ships with the CDK and is checked against its metaschema in unit
+        # tests, so a valid manifest skips `check_schema`. On failure the schema is checked first,
+        # so a broken schema is reported as `SchemaError` rather than blamed on the manifest.
+        # `best_match` keeps the error `jsonschema.validate` would raise.
+        validator_class = validator_for(self._declarative_component_schema)
         try:
-            validate(self._source_config, self._declarative_component_schema)
-        except ValidationError as e:
+            error = best_match(
+                validator_class(self._declarative_component_schema).iter_errors(self._source_config)
+            )
+        except Exception:
+            validator_class.check_schema(self._declarative_component_schema)
+            raise
+        if error is not None:
+            validator_class.check_schema(self._declarative_component_schema)
             raise ValidationError(
                 "Validation against json schema defined in declarative_component_schema.yaml schema failed"
-            ) from e
+            ) from error
 
     def _migrate_and_transform_config(
         self,

unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py

@@ -14,7 +14,8 @@
 import pytest
 import requests
 import yaml
-from jsonschema.exceptions import ValidationError
+from jsonschema.exceptions import SchemaError, UnknownType, ValidationError
+from jsonschema.validators import Draft7Validator, validate
 
 import unit_tests.sources.declarative.external_component  # Needed for dynamic imports to work
 from airbyte_cdk.legacy.sources.declarative.manifest_declarative_source import (
@@ -33,6 +34,7 @@
 )
 from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
     ConcurrentDeclarativeSource,
+    _get_declarative_component_schema,
 )
 from airbyte_cdk.sources.declarative.parsers.model_to_component_factory import (
     ModelToComponentFactory,
@@ -2540,3 +2542,105 @@ def capturing_resolve(resolver_self, *args, **kwargs):
         "HttpComponentsResolver's requester should have api_budget set during dynamic stream "
         "discovery, but it was None. This means discovery HTTP requests are not rate-limited."
     )
+
+
+def _two_stream_schema_validation_manifest() -> dict[str, Any]:
+    def stream(name: str) -> dict[str, Any]:
+        return {
+            "type": "DeclarativeStream",
+            "name": name,
+            "primary_key": [],
+            "schema_loader": {
+                "type": "InlineSchemaLoader",
+                "schema": {"type": "object", "properties": {}},
+            },
+            "retriever": {
+                "type": "SimpleRetriever",
+                "requester": {
+                    "type": "HttpRequester",
+                    "url_base": "https://api.test.com",
+                    "path": f"/{name}",
+                    "http_method": "GET",
+                },
+                "record_selector": {
+                    "type": "RecordSelector",
+                    "extractor": {"type": "DpathExtractor", "field_path": []},
+                },
+            },
+        }
+
+    return {
+        "version": "6.0.0",
+        "type": "DeclarativeSource",
+        "check": {"type": "CheckStream", "stream_names": ["a"]},
+        "streams": [stream("a"), stream("b")],
+    }
+
+
+def test_invalid_manifest_raises_the_best_match_error_not_the_first():
+    manifest = _two_stream_schema_validation_manifest()
+    manifest["streams"][0]["primary_key"] = 5
+    manifest["streams"][1]["retriever"]["requester"]["http_method"] = "FETCH"
+    schema = _get_declarative_component_schema()
+
+    assert next(Draft7Validator(schema).iter_errors(manifest)).json_path == "$.streams[0]"
+
+    with pytest.raises(ValidationError) as expected_error:
+        validate(manifest, schema)
+    expected = expected_error.value
+
+    with pytest.raises(ValidationError) as exc_info:
+        ManifestDeclarativeSource(source_config=manifest)
+
+    assert (
+        exc_info.value.message
+        == "Validation against json schema defined in declarative_component_schema.yaml schema failed"
+    )
+    cause = exc_info.value.__cause__
+    assert cause is not None
+    assert cause.message == expected.message
+    assert cause.json_path == expected.json_path
+    assert list(cause.schema_path) == list(expected.schema_path)
+    assert cause.validator == expected.validator
+    assert cause.validator_value == expected.validator_value
+    assert cause.json_path == "$.streams[1].retriever.requester.http_method"
+
+
+def test_building_a_source_does_not_check_the_component_schema():
+    with patch.object(
+        Draft7Validator,
+        "check_schema",
+        side_effect=AssertionError("check_schema must not run at source build"),
+    ):
+        ManifestDeclarativeSource(source_config=_two_stream_schema_validation_manifest())
+
+
+def test_a_broken_component_schema_that_fails_the_manifest_raises_schema_error():
+    manifest = _two_stream_schema_validation_manifest()
+    broken = deepcopy(_get_declarative_component_schema())
+    broken["definitions"]["HttpRequester"]["properties"]["http_method"]["enum"] = "GET"
+
+    assert list(Draft7Validator(broken).iter_errors(manifest))
+
+    with patch(
+        "airbyte_cdk.legacy.sources.declarative.manifest_declarative_source._get_declarative_component_schema",
+        return_value=broken,
+    ):
+        with pytest.raises(SchemaError):
+            ManifestDeclarativeSource(source_config=manifest)
+
+
+def test_a_broken_component_schema_that_breaks_validation_raises_schema_error():
+    manifest = _two_stream_schema_validation_manifest()
+    broken = deepcopy(_get_declarative_component_schema())
+    broken["definitions"]["HttpRequester"]["properties"]["path"]["type"] = "strng"
+
+    with pytest.raises(UnknownType):
+        list(Draft7Validator(broken).iter_errors(manifest))
+
+    with patch(
+        "airbyte_cdk.legacy.sources.declarative.manifest_declarative_source._get_declarative_component_schema",
+        return_value=broken,
+    ):
+        with pytest.raises(SchemaError):
+            ManifestDeclarativeSource(source_config=manifest)

unit_tests/sources/declarative/test_concurrent_declarative_source.py

@@ -25,7 +25,8 @@
     AirbyteTraceMessage,
     TraceType,
 )
-from jsonschema.exceptions import ValidationError
+from jsonschema.exceptions import SchemaError, UnknownType, ValidationError
+from jsonschema.validators import Draft7Validator, validate
 from typing_extensions import deprecated
 
 import unit_tests.sources.declarative.external_component  # Needed for dynamic imports to work
@@ -52,6 +53,7 @@
 from airbyte_cdk.sources.declarative.async_job.job_tracker import ConcurrentJobLimitReached
 from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
     ConcurrentDeclarativeSource,
+    _get_declarative_component_schema,
 )
 from airbyte_cdk.sources.declarative.extractors.record_filter import (
     ClientSideIncrementalRecordFilterDecorator,
@@ -6964,3 +6966,116 @@ def test_combined_extractor_skipping_empty_records_reads_every_page(
         6,
         7,
     ]
+
+
+def _two_stream_schema_validation_manifest() -> Dict[str, Any]:
+    def stream(name: str) -> Dict[str, Any]:
+        return {
+            "type": "DeclarativeStream",
+            "name": name,
+            "primary_key": [],
+            "schema_loader": {
+                "type": "InlineSchemaLoader",
+                "schema": {"type": "object", "properties": {}},
+            },
+            "retriever": {
+                "type": "SimpleRetriever",
+                "requester": {
+                    "type": "HttpRequester",
+                    "url_base": "https://api.test.com",
+                    "path": f"/{name}",
+                    "http_method": "GET",
+                },
+                "record_selector": {
+                    "type": "RecordSelector",
+                    "extractor": {"type": "DpathExtractor", "field_path": []},
+                },
+            },
+        }
+
+    return {
+        "version": "6.0.0",
+        "type": "DeclarativeSource",
+        "check": {"type": "CheckStream", "stream_names": ["a"]},
+        "streams": [stream("a"), stream("b")],
+    }
+
+
+def test_invalid_manifest_raises_the_best_match_error_not_the_first():
+    manifest = _two_stream_schema_validation_manifest()
+    manifest["streams"][0]["primary_key"] = 5
+    manifest["streams"][1]["retriever"]["requester"]["http_method"] = "FETCH"
+    schema = _get_declarative_component_schema()
+
+    assert next(Draft7Validator(schema).iter_errors(manifest)).json_path == "$.streams[0]"
+
+    with pytest.raises(ValidationError) as expected_error:
+        validate(manifest, schema)
+    expected = expected_error.value
+
+    with pytest.raises(ValidationError) as exc_info:
+        ConcurrentDeclarativeSource(
+            source_config=manifest, config={}, catalog=create_catalog("a"), state=None
+        )
+
+    assert (
+        exc_info.value.message
+        == "Validation against json schema defined in declarative_component_schema.yaml schema failed"
+    )
+    cause = exc_info.value.__cause__
+    assert cause is not None
+    assert cause.message == expected.message
+    assert cause.json_path == expected.json_path
+    assert list(cause.schema_path) == list(expected.schema_path)
+    assert cause.validator == expected.validator
+    assert cause.validator_value == expected.validator_value
+    assert cause.json_path == "$.streams[1].retriever.requester.http_method"
+
+
+def test_building_a_source_does_not_check_the_component_schema():
+    with patch.object(
+        Draft7Validator,
+        "check_schema",
+        side_effect=AssertionError("check_schema must not run at source build"),
+    ):
+        ConcurrentDeclarativeSource(
+            source_config=_two_stream_schema_validation_manifest(),
+            config={},
+            catalog=create_catalog("a"),
+            state=None,
+        )
+
+
+def test_a_broken_component_schema_that_fails_the_manifest_raises_schema_error():
+    manifest = _two_stream_schema_validation_manifest()
+    broken = copy.deepcopy(_get_declarative_component_schema())
+    broken["definitions"]["HttpRequester"]["properties"]["http_method"]["enum"] = "GET"
+
+    assert list(Draft7Validator(broken).iter_errors(manifest))
+
+    with patch(
+        "airbyte_cdk.sources.declarative.concurrent_declarative_source._get_declarative_component_schema",
+        return_value=broken,
+    ):
+        with pytest.raises(SchemaError):
+            ConcurrentDeclarativeSource(
+                source_config=manifest, config={}, catalog=create_catalog("a"), state=None
+            )
+
+
+def test_a_broken_component_schema_that_breaks_validation_raises_schema_error():
+    manifest = _two_stream_schema_validation_manifest()
+    broken = copy.deepcopy(_get_declarative_component_schema())
+    broken["definitions"]["HttpRequester"]["properties"]["path"]["type"] = "strng"
+
+    with pytest.raises(UnknownType):
+        list(Draft7Validator(broken).iter_errors(manifest))
+
+    with patch(
+        "airbyte_cdk.sources.declarative.concurrent_declarative_source._get_declarative_component_schema",
+        return_value=broken,
+    ):
+        with pytest.raises(SchemaError):
+            ConcurrentDeclarativeSource(
+                source_config=manifest, config={}, catalog=create_catalog("a"), state=None
+            )

unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py

@@ -22,6 +22,8 @@
 import pkgutil
 import re
 
+from jsonschema.validators import Draft7Validator, validator_for
+
 from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
     _get_declarative_component_schema,
 )
@@ -51,6 +53,14 @@ def test_the_shipped_component_schema_parses():
     assert "HttpRequester" in schema["definitions"]
 
 
+def test_the_shipped_component_schema_is_a_valid_draft_7_schema():
+    """`_validate_source` skips `check_schema` at runtime, so this is where a malformed schema is caught."""
+    schema = _get_declarative_component_schema()
+    validator_class = validator_for(schema)
+    assert validator_class is Draft7Validator
+    validator_class.check_schema(schema)
+
+
 def test_no_plain_scalar_carries_yaml_punctuation():
     """`": "` ends a plain scalar and `" #"` starts a comment, so either one truncates a value
     written as prose -- the first noisily, the second without a trace. Checked against the source