From 9fb976ce84dc02452f22765dea03397ef3cd612e Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 8 Oct 2026 22:18:56 +0000 Subject: [PATCH 1/9] fix(sdm): precompile dependency bytecode in the source-declarative-manifest image Co-Authored-By: bot_apk --- .github/workflows/docker-build-check.yml | 41 ++++++++++++++++++++++++ Dockerfile | 5 +++ 2 files changed, 46 insertions(+) diff --git a/.github/workflows/docker-build-check.yml b/.github/workflows/docker-build-check.yml index 0cc3036c4b..f20e6e8b85 100644 --- a/.github/workflows/docker-build-check.yml +++ b/.github/workflows/docker-build-check.yml @@ -45,6 +45,47 @@ jobs: tags: airbyte/source-declarative-manifest:pr-${{ github.event.pull_request.number }} outputs: type=image,name=target,annotation-index.org.opencontainers.image.description=SDM Docker image for PR ${{ github.event.pull_request.number }} + - name: Build Docker image for bytecode check (linux/amd64) + uses: docker/build-push-action@v5 + with: + context: . + platforms: linux/amd64 + load: true + tags: airbyte/source-declarative-manifest:bytecode-check + + - name: Verify every site-packages module has precompiled bytecode + run: | + docker run --rm -i --entrypoint python airbyte/source-declarative-manifest:bytecode-check - <<'EOF' + import importlib.util + import pathlib + import struct + import sys + import sysconfig + + root = pathlib.Path(sysconfig.get_paths()["purelib"]) + missing, stale = [], [] + total = 0 + for src in root.rglob("*.py"): + total += 1 + pyc = pathlib.Path(importlib.util.cache_from_source(str(src))) + if not pyc.is_file(): + missing.append(src) + continue + header = pyc.read_bytes()[:16] + flags = int.from_bytes(header[4:8], "little") + if header[:4] != importlib.util.MAGIC_NUMBER: + stale.append(src) + elif flags == 0: + mtime, size = struct.unpack(" main.py \ From 9a61b6d1647a2ef1e5ddeb62c3235227aada9e9c Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 8 Oct 2026 23:00:24 +0000 Subject: [PATCH 2/9] perf(low-code): parse manifest and component schema with libyaml CSafeLoader when available Co-Authored-By: bot_apk --- .../cli/source_declarative_manifest/_run.py | 4 +- .../manifest_declarative_source.py | 4 +- .../concurrent_declarative_source.py | 4 +- .../declarative/yaml_declarative_source.py | 7 +- airbyte_cdk/utils/yaml_loader.py | 17 +++ unit_tests/utils/test_yaml_loader.py | 117 ++++++++++++++++++ 6 files changed, 143 insertions(+), 10 deletions(-) create mode 100644 airbyte_cdk/utils/yaml_loader.py create mode 100644 unit_tests/utils/test_yaml_loader.py diff --git a/airbyte_cdk/cli/source_declarative_manifest/_run.py b/airbyte_cdk/cli/source_declarative_manifest/_run.py index 1b64e51252..13b13104bf 100644 --- a/airbyte_cdk/cli/source_declarative_manifest/_run.py +++ b/airbyte_cdk/cli/source_declarative_manifest/_run.py @@ -26,7 +26,6 @@ from typing import Any, cast import orjson -import yaml from airbyte_cdk.entrypoint import AirbyteEntrypoint, launch from airbyte_cdk.models import ( @@ -50,6 +49,7 @@ from airbyte_cdk.sources.declarative.yaml_declarative_source import YamlDeclarativeSource from airbyte_cdk.sources.source import TState from airbyte_cdk.utils.datetime_helpers import ab_datetime_now +from airbyte_cdk.utils.yaml_loader import safe_load_yaml class SourceLocalYaml(YamlDeclarativeSource): @@ -272,7 +272,7 @@ def _parse_manifest_from_file(filepath: str) -> dict[str, Any] | None: """Extract and parse a manifest file specified in the args.""" try: with open(filepath, "r", encoding="utf-8") as manifest_file: - manifest_content = yaml.safe_load(manifest_file) + manifest_content = safe_load_yaml(manifest_file) if manifest_content is None: raise ValueError(f"Manifest file at {filepath} is empty") if not isinstance(manifest_content, dict): diff --git a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py index 5a20bdf1f3..2c6d5c8bd2 100644 --- a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py +++ b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py @@ -11,7 +11,6 @@ from typing import Any, Dict, Iterator, List, Mapping, Optional, Set, Union import orjson -import yaml from jsonschema.exceptions import ValidationError from jsonschema.validators import validate from packaging.version import InvalidVersion, Version @@ -75,6 +74,7 @@ SliceLogger, ) from airbyte_cdk.utils.traced_exception import AirbyteTracedException +from airbyte_cdk.utils.yaml_loader import safe_load_yaml def _get_declarative_component_schema() -> Dict[str, Any]: @@ -83,7 +83,7 @@ def _get_declarative_component_schema() -> Dict[str, Any]: "airbyte_cdk", "sources/declarative/declarative_component_schema.yaml" ) if raw_component_schema is not None: - declarative_component_schema = yaml.load(raw_component_schema, Loader=yaml.SafeLoader) + declarative_component_schema = safe_load_yaml(raw_component_schema) return declarative_component_schema # type: ignore else: raise RuntimeError( diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index 055576f45f..1fafa07a62 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -20,7 +20,6 @@ ) import orjson -import yaml from airbyte_protocol_dataclasses.models import AirbyteStreamStatus, Level, StreamDescriptor from jsonschema.exceptions import ValidationError from jsonschema.validators import validate @@ -105,6 +104,7 @@ from airbyte_cdk.utils.airbyte_secrets_utils import add_to_secrets, get_secrets from airbyte_cdk.utils.stream_status_utils import as_airbyte_message from airbyte_cdk.utils.traced_exception import AirbyteTracedException +from airbyte_cdk.utils.yaml_loader import safe_load_yaml @dataclass @@ -128,7 +128,7 @@ def _get_declarative_component_schema() -> Dict[str, Any]: "airbyte_cdk", "sources/declarative/declarative_component_schema.yaml" ) if raw_component_schema is not None: - declarative_component_schema = yaml.load(raw_component_schema, Loader=yaml.SafeLoader) + declarative_component_schema = safe_load_yaml(raw_component_schema) return declarative_component_schema # type: ignore else: raise RuntimeError( diff --git a/airbyte_cdk/sources/declarative/yaml_declarative_source.py b/airbyte_cdk/sources/declarative/yaml_declarative_source.py index 7af07a5c8f..c397e7b1cc 100644 --- a/airbyte_cdk/sources/declarative/yaml_declarative_source.py +++ b/airbyte_cdk/sources/declarative/yaml_declarative_source.py @@ -5,13 +5,12 @@ import pkgutil from typing import Any, List, Mapping, Optional -import yaml - from airbyte_cdk.models import AirbyteStateMessage, ConfiguredAirbyteCatalog from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( ConcurrentDeclarativeSource, ) from airbyte_cdk.sources.types import ConnectionDefinition +from airbyte_cdk.utils.yaml_loader import safe_load_yaml class YamlDeclarativeSource(ConcurrentDeclarativeSource): @@ -44,7 +43,7 @@ def _read_and_parse_yaml_file(self, path_to_yaml_file: str) -> ConnectionDefinit try: # For testing purposes, we want to allow to just pass a file with open(path_to_yaml_file, "r") as f: - return yaml.safe_load(f) # type: ignore # we assume the yaml represents a ConnectionDefinition + return safe_load_yaml(f) # type: ignore # we assume the yaml represents a ConnectionDefinition except FileNotFoundError: # Running inside the container, the working directory during an operation is not structured the same as the static files package = self.__class__.__module__.split(".")[0] @@ -66,4 +65,4 @@ def _parse(connection_definition_str: str) -> ConnectionDefinition: :param connection_definition_str: yaml string to parse :return: The ConnectionDefinition parsed from connection_definition_str """ - return yaml.safe_load(connection_definition_str) # type: ignore # yaml.safe_load doesn't return a type but know it is a Mapping + return safe_load_yaml(connection_definition_str) # type: ignore # safe_load_yaml doesn't return a type but know it is a Mapping diff --git a/airbyte_cdk/utils/yaml_loader.py b/airbyte_cdk/utils/yaml_loader.py new file mode 100644 index 0000000000..382212e29b --- /dev/null +++ b/airbyte_cdk/utils/yaml_loader.py @@ -0,0 +1,17 @@ +# +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +# + +from typing import IO, Any, Union + +import yaml + + +def get_safe_loader() -> Any: + """Return libyaml's CSafeLoader when PyYAML was built with it, else the pure-Python SafeLoader.""" + return getattr(yaml, "CSafeLoader", yaml.SafeLoader) + + +def safe_load_yaml(stream: Union[str, bytes, IO[str], IO[bytes]]) -> Any: + """Equivalent to yaml.safe_load, but uses the C loader when available.""" + return yaml.load(stream, Loader=get_safe_loader()) diff --git a/unit_tests/utils/test_yaml_loader.py b/unit_tests/utils/test_yaml_loader.py new file mode 100644 index 0000000000..fceb89e014 --- /dev/null +++ b/unit_tests/utils/test_yaml_loader.py @@ -0,0 +1,117 @@ +# +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +# + +import pkgutil +from pathlib import Path +from typing import Any + +import pytest +import yaml + +from airbyte_cdk.cli.source_declarative_manifest._run import _parse_manifest_from_file +from airbyte_cdk.legacy.sources.declarative.manifest_declarative_source import ( + _get_declarative_component_schema as _legacy_get_declarative_component_schema, +) +from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( + _get_declarative_component_schema, +) +from airbyte_cdk.sources.declarative.yaml_declarative_source import YamlDeclarativeSource +from airbyte_cdk.utils.yaml_loader import get_safe_loader, safe_load_yaml + +requires_libyaml = pytest.mark.skipif( + not yaml.__with_libyaml__, reason="PyYAML was built without libyaml" +) + +MALFORMED_YAML = """ +version: "version" +definitions: + this is not parsable yaml: " at all +streams: + - type: DeclarativeStream + $parameters: + name: "lists" + primary_key: id + url_base: "https://api.sendgrid.com" +check: + type: CheckStream + stream_names: ["lists"] +""" + + +@requires_libyaml +def test_get_safe_loader_uses_c_loader_when_available() -> None: + assert get_safe_loader() is yaml.CSafeLoader + + +def test_falls_back_to_pure_python_loader_without_libyaml(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delattr(yaml, "CSafeLoader", raising=False) + + assert get_safe_loader() is yaml.SafeLoader + assert safe_load_yaml("a: 1\nb: [x, y]") == {"a": 1, "b": ["x", "y"]} + assert YamlDeclarativeSource._parse("version: 1.0.0\nstreams: []") == { + "version": "1.0.0", + "streams": [], + } + + +@requires_libyaml +def test_component_schema_parses_identically_with_both_loaders() -> None: + raw = pkgutil.get_data("airbyte_cdk", "sources/declarative/declarative_component_schema.yaml") + assert raw is not None + pure_python_schema = yaml.load(raw, Loader=yaml.SafeLoader) + c_loader_schema = yaml.load(raw, Loader=yaml.CSafeLoader) + assert pure_python_schema == c_loader_schema + assert c_loader_schema == _get_declarative_component_schema() + + +@pytest.mark.parametrize( + "loader", + [ + yaml.SafeLoader, + pytest.param( + yaml.CSafeLoader, + marks=pytest.mark.skipif( + not yaml.__with_libyaml__, reason="PyYAML was built without libyaml" + ), + ), + ], +) +def test_malformed_yaml_raises_parser_error_with_both_loaders(loader: Any) -> None: + with pytest.raises(yaml.parser.ParserError): + yaml.load(MALFORMED_YAML, Loader=loader) + + +@requires_libyaml +def test_declarative_call_sites_use_c_loader( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + instantiated: list[Any] = [] + + class _Spy(yaml.CSafeLoader): + def __init__(self, stream: Any) -> None: + instantiated.append(stream) + super().__init__(stream) + + monkeypatch.setattr(yaml, "CSafeLoader", _Spy) + + _get_declarative_component_schema() + assert len(instantiated) == 1 + + _legacy_get_declarative_component_schema() + assert len(instantiated) == 2 + + YamlDeclarativeSource._parse("a: 1") + assert len(instantiated) == 3 + + yaml_file = tmp_path / "source.yaml" + yaml_file.write_text("a: 1") + YamlDeclarativeSource._read_and_parse_yaml_file( + YamlDeclarativeSource.__new__(YamlDeclarativeSource), str(yaml_file) + ) + assert len(instantiated) == 4 + + manifest_file = tmp_path / "m.yaml" + manifest_file.write_text("version: 1.0.0\nstreams: []") + _parse_manifest_from_file(str(manifest_file)) + assert len(instantiated) == 5 From ff64dbecb75d9947600ca3852a4c3e80af0e27ec Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 8 Oct 2026 23:42:24 +0000 Subject: [PATCH 3/9] perf(declarative): skip runtime metaschema check of the component schema Co-Authored-By: bot_apk --- .../manifest_declarative_source.py | 17 ++-- .../concurrent_declarative_source.py | 17 ++-- .../test_manifest_declarative_source.py | 73 +++++++++++++++++ .../test_concurrent_declarative_source.py | 80 +++++++++++++++++++ ...eclarative_component_schema_is_loadable.py | 10 +++ 5 files changed, 185 insertions(+), 12 deletions(-) diff --git a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py index 5a20bdf1f3..46b0afb905 100644 --- a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py +++ b/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,17 @@ def _validate_source(self) -> None: Validates the connector manifest against the declarative component schema """ - try: - validate(self._source_config, self._declarative_component_schema) - except ValidationError as e: + # The component schema ships with the CDK and is checked against its metaschema in unit + # tests, so `jsonschema.validate`'s `check_schema` step is skipped. `best_match` keeps the + # error `validate` would raise. + validator = validator_for(self._declarative_component_schema)( + self._declarative_component_schema + ) + error = best_match(validator.iter_errors(self._source_config)) + if error is not None: 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") diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index 055576f45f..6ebb677f3d 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/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,17 @@ def _validate_source(self) -> None: Validates the connector manifest against the declarative component schema """ - try: - validate(self._source_config, self._declarative_component_schema) - except ValidationError as e: + # The component schema ships with the CDK and is checked against its metaschema in unit + # tests, so `jsonschema.validate`'s `check_schema` step is skipped. `best_match` keeps the + # error `validate` would raise. + validator = validator_for(self._declarative_component_schema)( + self._declarative_component_schema + ) + error = best_match(validator.iter_errors(self._source_config)) + if error is not None: raise ValidationError( "Validation against json schema defined in declarative_component_schema.yaml schema failed" - ) from e + ) from error def _migrate_and_transform_config( self, diff --git a/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py b/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py index b565874323..c69c2d8677 100644 --- a/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py +++ b/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py @@ -15,6 +15,7 @@ import requests import yaml from jsonschema.exceptions import 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,74 @@ 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()) diff --git a/unit_tests/sources/declarative/test_concurrent_declarative_source.py b/unit_tests/sources/declarative/test_concurrent_declarative_source.py index 686bd54f0b..37c67b1b54 100644 --- a/unit_tests/sources/declarative/test_concurrent_declarative_source.py +++ b/unit_tests/sources/declarative/test_concurrent_declarative_source.py @@ -26,6 +26,7 @@ TraceType, ) from jsonschema.exceptions import 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,81 @@ 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, + ) diff --git a/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py index db7d849bba..3f8dee86b9 100644 --- a/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py +++ b/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 From 4864013c4fff769375daa53af97195568e57b071 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 9 Oct 2026 17:13:57 +0000 Subject: [PATCH 4/9] perf(low-code): keep pure-Python YAML error messages when the C loader fails Co-Authored-By: bot_apk --- airbyte_cdk/utils/yaml_loader.py | 37 ++++++++++- unit_tests/utils/test_yaml_loader.py | 95 ++++++++++++++++++++++++++++ 2 files changed, 129 insertions(+), 3 deletions(-) diff --git a/airbyte_cdk/utils/yaml_loader.py b/airbyte_cdk/utils/yaml_loader.py index 382212e29b..bbc6a37a76 100644 --- a/airbyte_cdk/utils/yaml_loader.py +++ b/airbyte_cdk/utils/yaml_loader.py @@ -2,7 +2,7 @@ # Copyright (c) 2025 Airbyte, Inc., all rights reserved. # -from typing import IO, Any, Union +from typing import IO, Any, Optional, Union, cast import yaml @@ -12,6 +12,37 @@ def get_safe_loader() -> Any: return getattr(yaml, "CSafeLoader", yaml.SafeLoader) +def _tell_if_seekable(stream: Any) -> Optional[int]: + """Return the stream's current position if it can be rewound, else None.""" + if isinstance(stream, (str, bytes)): + return None + seekable = getattr(stream, "seekable", None) + if not callable(seekable) or not seekable(): + return None + try: + return cast(int, stream.tell()) + except (OSError, ValueError): + return None + + def safe_load_yaml(stream: Union[str, bytes, IO[str], IO[bytes]]) -> Any: - """Equivalent to yaml.safe_load, but uses the C loader when available.""" - return yaml.load(stream, Loader=get_safe_loader()) + """Equivalent to yaml.safe_load, but uses the C loader when available. + + If the C loader raises a yaml.YAMLError, the input is re-parsed with the + pure-Python SafeLoader and that error is raised instead. The C loader's + messages lack the source-line snippet and differ in wording, so the re-parse + preserves the error messages users saw before the C loader was introduced. + """ + loader = get_safe_loader() + if loader is yaml.SafeLoader: + return yaml.load(stream, Loader=yaml.SafeLoader) + rewind_to = _tell_if_seekable(stream) + try: + return yaml.load(stream, Loader=loader) + except yaml.YAMLError: + if isinstance(stream, (str, bytes)): + return yaml.load(stream, Loader=yaml.SafeLoader) + if rewind_to is None: + raise + stream.seek(rewind_to) + return yaml.load(stream, Loader=yaml.SafeLoader) diff --git a/unit_tests/utils/test_yaml_loader.py b/unit_tests/utils/test_yaml_loader.py index fceb89e014..5819ccccbb 100644 --- a/unit_tests/utils/test_yaml_loader.py +++ b/unit_tests/utils/test_yaml_loader.py @@ -2,6 +2,7 @@ # Copyright (c) 2025 Airbyte, Inc., all rights reserved. # +import io import pkgutil from pathlib import Path from typing import Any @@ -115,3 +116,97 @@ def __init__(self, stream: Any) -> None: manifest_file.write_text("version: 1.0.0\nstreams: []") _parse_manifest_from_file(str(manifest_file)) assert len(instantiated) == 5 + + +# A list item whose nested mapping is indented one space short. +MISINDENTED_YAML = ( + 'version: "1"\nstreams:\n - type: DeclarativeStream\n name: a\n primary_key: id\n' +) + + +@requires_libyaml +def test_syntax_error_in_string_matches_pure_python_message() -> None: + with pytest.raises(yaml.YAMLError) as expected_exc_info: + yaml.load(MISINDENTED_YAML, Loader=yaml.SafeLoader) + + with pytest.raises(yaml.YAMLError) as actual_exc_info: + safe_load_yaml(MISINDENTED_YAML) + + assert str(actual_exc_info.value) == str(expected_exc_info.value) + assert type(actual_exc_info.value) is type(expected_exc_info.value) + assert "^" in str(actual_exc_info.value) + + with pytest.raises(yaml.YAMLError) as parse_exc_info: + YamlDeclarativeSource._parse(MISINDENTED_YAML) + + assert str(parse_exc_info.value) == str(expected_exc_info.value) + assert type(parse_exc_info.value) is type(expected_exc_info.value) + + +@requires_libyaml +def test_syntax_error_in_file_matches_pure_python_message(tmp_path: Path) -> None: + path = tmp_path / "manifest.yaml" + path.write_text(MISINDENTED_YAML) + + with open(path) as f: + with pytest.raises(yaml.YAMLError) as expected_exc_info: + yaml.load(f, Loader=yaml.SafeLoader) + + with open(path) as f: + with pytest.raises(yaml.YAMLError) as actual_exc_info: + safe_load_yaml(f) + + assert str(actual_exc_info.value) == str(expected_exc_info.value) + assert str(path) in str(actual_exc_info.value) + + +@requires_libyaml +def test_syntax_error_in_non_seekable_stream_keeps_c_loader_message() -> None: + class _NonSeekable(io.StringIO): + def seekable(self) -> bool: + return False + + with pytest.raises(yaml.YAMLError) as expected_exc_info: + yaml.load(_NonSeekable(MISINDENTED_YAML), Loader=yaml.CSafeLoader) + + with pytest.raises(yaml.YAMLError) as actual_exc_info: + safe_load_yaml(_NonSeekable(MISINDENTED_YAML)) + + assert str(actual_exc_info.value) == str(expected_exc_info.value) + + +@requires_libyaml +def test_successful_parse_never_uses_pure_python_loader( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + instantiated: list[Any] = [] + + class _Spy(yaml.SafeLoader): + def __init__(self, stream: Any) -> None: + instantiated.append(stream) + super().__init__(stream) + + monkeypatch.setattr(yaml, "SafeLoader", _Spy) + + assert safe_load_yaml("a: 1") == {"a": 1} + assert safe_load_yaml(b"a: 1") == {"a": 1} + + yaml_file = tmp_path / "source.yaml" + yaml_file.write_text("a: 1") + with open(yaml_file) as f: + assert safe_load_yaml(f) == {"a": 1} + + assert instantiated == [] + + +@requires_libyaml +def test_pure_python_result_returned_if_only_c_loader_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class _FailingLoader(yaml.CSafeLoader): + def get_single_data(self) -> Any: + raise yaml.YAMLError("boom") + + monkeypatch.setattr(yaml, "CSafeLoader", _FailingLoader) + + assert safe_load_yaml("a: 1") == {"a": 1} From 29b03be065320bce03556a3326b00b4f5d7a1df0 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 9 Oct 2026 17:15:25 +0000 Subject: [PATCH 5/9] perf(declarative): check the component schema when manifest validation fails Co-Authored-By: bot_apk --- .../manifest_declarative_source.py | 18 ++++++--- .../concurrent_declarative_source.py | 18 ++++++--- .../test_manifest_declarative_source.py | 33 ++++++++++++++++- .../test_concurrent_declarative_source.py | 37 ++++++++++++++++++- 4 files changed, 92 insertions(+), 14 deletions(-) diff --git a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py index 46b0afb905..b4a92e0c7f 100644 --- a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py +++ b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py @@ -448,13 +448,19 @@ def _validate_source(self) -> None: """ # The component schema ships with the CDK and is checked against its metaschema in unit - # tests, so `jsonschema.validate`'s `check_schema` step is skipped. `best_match` keeps the - # error `validate` would raise. - validator = validator_for(self._declarative_component_schema)( - self._declarative_component_schema - ) - error = best_match(validator.iter_errors(self._source_config)) + # 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: + 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 error diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index 6ebb677f3d..5db724100e 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -346,13 +346,19 @@ def _validate_source(self) -> None: """ # The component schema ships with the CDK and is checked against its metaschema in unit - # tests, so `jsonschema.validate`'s `check_schema` step is skipped. `best_match` keeps the - # error `validate` would raise. - validator = validator_for(self._declarative_component_schema)( - self._declarative_component_schema - ) - error = best_match(validator.iter_errors(self._source_config)) + # 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: + 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 error diff --git a/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py b/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py index c69c2d8677..58dce25e5b 100644 --- a/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py +++ b/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py @@ -14,7 +14,7 @@ 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 @@ -2613,3 +2613,34 @@ def test_building_a_source_does_not_check_the_component_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) diff --git a/unit_tests/sources/declarative/test_concurrent_declarative_source.py b/unit_tests/sources/declarative/test_concurrent_declarative_source.py index 37c67b1b54..5718d072d7 100644 --- a/unit_tests/sources/declarative/test_concurrent_declarative_source.py +++ b/unit_tests/sources/declarative/test_concurrent_declarative_source.py @@ -25,7 +25,7 @@ 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 @@ -7044,3 +7044,38 @@ def test_building_a_source_does_not_check_the_component_schema(): 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 + ) From c24a48bdcf089306f6e9b18992c22352adea7000 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 21 Oct 2026 19:31:03 +0000 Subject: [PATCH 6/9] fix(low-code): address review on YAML C loader fallback Co-Authored-By: bot_apk --- airbyte_cdk/utils/yaml_loader.py | 55 +++++++++------- unit_tests/utils/test_yaml_loader.py | 94 +++++++++++++++++++--------- 2 files changed, 94 insertions(+), 55 deletions(-) diff --git a/airbyte_cdk/utils/yaml_loader.py b/airbyte_cdk/utils/yaml_loader.py index bbc6a37a76..6ef5e60834 100644 --- a/airbyte_cdk/utils/yaml_loader.py +++ b/airbyte_cdk/utils/yaml_loader.py @@ -2,47 +2,54 @@ # Copyright (c) 2025 Airbyte, Inc., all rights reserved. # +from __future__ import annotations + from typing import IO, Any, Optional, Union, cast import yaml -def get_safe_loader() -> Any: +def _get_safe_loader() -> Union[type[yaml.SafeLoader], type[yaml.CSafeLoader]]: """Return libyaml's CSafeLoader when PyYAML was built with it, else the pure-Python SafeLoader.""" - return getattr(yaml, "CSafeLoader", yaml.SafeLoader) + # The annotations stay unevaluated: yaml.CSafeLoader does not exist without libyaml. + return cast( + "Union[type[yaml.SafeLoader], type[yaml.CSafeLoader]]", + getattr(yaml, "CSafeLoader", yaml.SafeLoader), + ) -def _tell_if_seekable(stream: Any) -> Optional[int]: +def _tell_if_seekable(stream: Union[IO[str], IO[bytes]]) -> Optional[int]: """Return the stream's current position if it can be rewound, else None.""" - if isinstance(stream, (str, bytes)): - return None - seekable = getattr(stream, "seekable", None) - if not callable(seekable) or not seekable(): - return None try: - return cast(int, stream.tell()) - except (OSError, ValueError): + if stream.seekable() is not True: + return None + position = stream.tell() + except (AttributeError, OSError, ValueError): return None + return position if isinstance(position, int) else None def safe_load_yaml(stream: Union[str, bytes, IO[str], IO[bytes]]) -> Any: - """Equivalent to yaml.safe_load, but uses the C loader when available. + """Parse trusted YAML like yaml.safe_load, using libyaml's CSafeLoader when available. - If the C loader raises a yaml.YAMLError, the input is re-parsed with the - pure-Python SafeLoader and that error is raised instead. The C loader's - messages lack the source-line snippet and differ in wording, so the re-parse - preserves the error messages users saw before the C loader was introduced. + Results match yaml.safe_load for documents both loaders accept. libyaml accepts + some input SafeLoader rejects (for example a tab after `key:`), ignores constructors + and resolvers registered on yaml.SafeLoader, and can crash the process on very deeply + nested input, so do not use this for untrusted input. When libyaml raises a + YAMLError, the input is parsed again with SafeLoader so callers see SafeLoader's error. """ - loader = get_safe_loader() - if loader is yaml.SafeLoader: + loader = _get_safe_loader() + exact_text = type(stream) in (str, bytes) + rewind_to = None + if not exact_text and not isinstance(stream, (str, bytes)): + rewind_to = _tell_if_seekable(stream) + # str/bytes subclasses, non-seekable streams and mocks keep today's behaviour. + if loader is yaml.SafeLoader or (not exact_text and rewind_to is None): return yaml.load(stream, Loader=yaml.SafeLoader) - rewind_to = _tell_if_seekable(stream) try: return yaml.load(stream, Loader=loader) except yaml.YAMLError: - if isinstance(stream, (str, bytes)): - return yaml.load(stream, Loader=yaml.SafeLoader) - if rewind_to is None: - raise - stream.seek(rewind_to) - return yaml.load(stream, Loader=yaml.SafeLoader) + pass # Re-parse outside the handler so libyaml's error is not chained. + if rewind_to is not None: + cast(Union[IO[str], IO[bytes]], stream).seek(rewind_to) + return yaml.load(stream, Loader=yaml.SafeLoader) diff --git a/unit_tests/utils/test_yaml_loader.py b/unit_tests/utils/test_yaml_loader.py index 5819ccccbb..13ee02e173 100644 --- a/unit_tests/utils/test_yaml_loader.py +++ b/unit_tests/utils/test_yaml_loader.py @@ -6,6 +6,7 @@ import pkgutil from pathlib import Path from typing import Any +from unittest import mock import pytest import yaml @@ -18,7 +19,7 @@ _get_declarative_component_schema, ) from airbyte_cdk.sources.declarative.yaml_declarative_source import YamlDeclarativeSource -from airbyte_cdk.utils.yaml_loader import get_safe_loader, safe_load_yaml +from airbyte_cdk.utils.yaml_loader import _get_safe_loader, safe_load_yaml requires_libyaml = pytest.mark.skipif( not yaml.__with_libyaml__, reason="PyYAML was built without libyaml" @@ -39,16 +40,28 @@ stream_names: ["lists"] """ +# A list item whose nested mapping is indented one space short. +MISINDENTED_YAML = ( + 'version: "1"\nstreams:\n - type: DeclarativeStream\n name: a\n primary_key: id\n' +) + +ESCAPED_EMOJI_YAML = 'description: "\\ud83d\\ude00 emoji"\n' + + +class _NonSeekable(io.StringIO): + def seekable(self) -> bool: + return False + @requires_libyaml def test_get_safe_loader_uses_c_loader_when_available() -> None: - assert get_safe_loader() is yaml.CSafeLoader + assert _get_safe_loader() is yaml.CSafeLoader def test_falls_back_to_pure_python_loader_without_libyaml(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delattr(yaml, "CSafeLoader", raising=False) - assert get_safe_loader() is yaml.SafeLoader + assert _get_safe_loader() is yaml.SafeLoader assert safe_load_yaml("a: 1\nb: [x, y]") == {"a": 1, "b": ["x", "y"]} assert YamlDeclarativeSource._parse("version: 1.0.0\nstreams: []") == { "version": "1.0.0", @@ -63,24 +76,22 @@ def test_component_schema_parses_identically_with_both_loaders() -> None: pure_python_schema = yaml.load(raw, Loader=yaml.SafeLoader) c_loader_schema = yaml.load(raw, Loader=yaml.CSafeLoader) assert pure_python_schema == c_loader_schema + assert repr(pure_python_schema) == repr(c_loader_schema) assert c_loader_schema == _get_declarative_component_schema() -@pytest.mark.parametrize( - "loader", - [ - yaml.SafeLoader, - pytest.param( - yaml.CSafeLoader, - marks=pytest.mark.skipif( - not yaml.__with_libyaml__, reason="PyYAML was built without libyaml" - ), - ), - ], -) -def test_malformed_yaml_raises_parser_error_with_both_loaders(loader: Any) -> None: +@requires_libyaml +def test_malformed_yaml_raises_parser_error_with_c_loader() -> None: with pytest.raises(yaml.parser.ParserError): - yaml.load(MALFORMED_YAML, Loader=loader) + safe_load_yaml(MALFORMED_YAML) + + +def test_malformed_yaml_raises_parser_error_without_libyaml( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delattr(yaml, "CSafeLoader", raising=False) + with pytest.raises(yaml.parser.ParserError): + safe_load_yaml(MALFORMED_YAML) @requires_libyaml @@ -118,12 +129,6 @@ def __init__(self, stream: Any) -> None: assert len(instantiated) == 5 -# A list item whose nested mapping is indented one space short. -MISINDENTED_YAML = ( - 'version: "1"\nstreams:\n - type: DeclarativeStream\n name: a\n primary_key: id\n' -) - - @requires_libyaml def test_syntax_error_in_string_matches_pure_python_message() -> None: with pytest.raises(yaml.YAMLError) as expected_exc_info: @@ -134,6 +139,7 @@ def test_syntax_error_in_string_matches_pure_python_message() -> None: assert str(actual_exc_info.value) == str(expected_exc_info.value) assert type(actual_exc_info.value) is type(expected_exc_info.value) + assert actual_exc_info.value.__context__ is None assert "^" in str(actual_exc_info.value) with pytest.raises(yaml.YAMLError) as parse_exc_info: @@ -141,6 +147,7 @@ def test_syntax_error_in_string_matches_pure_python_message() -> None: assert str(parse_exc_info.value) == str(expected_exc_info.value) assert type(parse_exc_info.value) is type(expected_exc_info.value) + assert parse_exc_info.value.__context__ is None @requires_libyaml @@ -157,22 +164,47 @@ def test_syntax_error_in_file_matches_pure_python_message(tmp_path: Path) -> Non safe_load_yaml(f) assert str(actual_exc_info.value) == str(expected_exc_info.value) + assert actual_exc_info.value.__context__ is None assert str(path) in str(actual_exc_info.value) @requires_libyaml -def test_syntax_error_in_non_seekable_stream_keeps_c_loader_message() -> None: - class _NonSeekable(io.StringIO): - def seekable(self) -> bool: - return False +def test_syntax_error_does_not_chain_c_loader_error() -> None: + with pytest.raises(yaml.YAMLError) as exc_info: + safe_load_yaml(MISINDENTED_YAML) + assert exc_info.value.__context__ is None - with pytest.raises(yaml.YAMLError) as expected_exc_info: - yaml.load(_NonSeekable(MISINDENTED_YAML), Loader=yaml.CSafeLoader) - with pytest.raises(yaml.YAMLError) as actual_exc_info: +@requires_libyaml +def test_non_seekable_stream_parses_and_errors_like_safe_loader() -> None: + assert safe_load_yaml(_NonSeekable(ESCAPED_EMOJI_YAML)) == yaml.load( + _NonSeekable(ESCAPED_EMOJI_YAML), Loader=yaml.SafeLoader + ) + with pytest.raises(yaml.YAMLError) as expected: + yaml.load(_NonSeekable(MISINDENTED_YAML), Loader=yaml.SafeLoader) + with pytest.raises(yaml.YAMLError) as actual: safe_load_yaml(_NonSeekable(MISINDENTED_YAML)) + assert str(actual.value) == str(expected.value) - assert str(actual_exc_info.value) == str(expected_exc_info.value) + +@requires_libyaml +def test_mock_open_handle_with_invalid_yaml_raises() -> None: + with pytest.raises(yaml.YAMLError): + safe_load_yaml(mock.mock_open(read_data=MISINDENTED_YAML)()) + + +def test_str_subclass_parses() -> None: + class _Str(str): + pass + + assert safe_load_yaml(_Str("a: 1")) == {"a": 1} + + +@requires_libyaml +def test_c_loader_accepts_tab_separators_that_safe_loader_rejects() -> None: + with pytest.raises(yaml.scanner.ScannerError): + yaml.load("a:\tb", Loader=yaml.SafeLoader) + assert safe_load_yaml("a:\tb") == {"a": "b"} @requires_libyaml From 54d193a41d83754a0e349fd775a3b1fbc45794bf Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 21 Oct 2026 19:35:36 +0000 Subject: [PATCH 7/9] refactor(declarative): share manifest validation and tighten its tests Co-Authored-By: bot_apk --- .../manifest_declarative_source.py | 22 +--- .../concurrent_declarative_source.py | 39 +++--- .../test_manifest_declarative_source.py | 116 ++++-------------- .../test_concurrent_declarative_source.py | 41 +++++-- ...eclarative_component_schema_is_loadable.py | 11 +- 5 files changed, 86 insertions(+), 143 deletions(-) diff --git a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py index b4a92e0c7f..b8d6f5571b 100644 --- a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py +++ b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py @@ -12,8 +12,7 @@ import orjson import yaml -from jsonschema.exceptions import ValidationError, best_match -from jsonschema.validators import validator_for +from jsonschema.exceptions import ValidationError from packaging.version import InvalidVersion, Version from airbyte_cdk.config_observation import create_connector_config_control_message @@ -35,6 +34,7 @@ from airbyte_cdk.models.airbyte_protocol_serializers import AirbyteMessageSerializer from airbyte_cdk.sources.declarative.checks import COMPONENTS_CHECKER_TYPE_MAPPING from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker +from airbyte_cdk.sources.declarative.concurrent_declarative_source import _validate_manifest from airbyte_cdk.sources.declarative.interpolation import InterpolatedBoolean from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( ConditionalStreams as ConditionalStreamsModel, @@ -447,23 +447,7 @@ 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: - 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 error + _validate_manifest(self._source_config, self._declarative_component_schema) cdk_version_str = metadata.version("airbyte_cdk") cdk_version = self._parse_version(cdk_version_str, "airbyte-cdk") diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index 5db724100e..cc4688921b 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -140,6 +140,27 @@ def _get_declarative_component_schema() -> Dict[str, Any]: ) +def _validate_manifest(manifest: Mapping[str, Any], schema: Mapping[str, Any]) -> None: + # The component schema ships with the CDK and is checked against its metaschema in unit + # tests, so a valid manifest skips `check_schema`. If validation fails or raises, + # `check_schema` runs before the error propagates, so a broken schema surfaces as + # `SchemaError` instead of being blamed on the manifest. `best_match` keeps the error + # `jsonschema.validate` would raise; raising it here gives the cause a traceback. + validator_class = validator_for(schema) + try: + error = best_match(validator_class(schema).iter_errors(manifest)) + if error is not None: + raise error + except ValidationError as e: + validator_class.check_schema(schema) + raise ValidationError( + "Validation against json schema defined in declarative_component_schema.yaml schema failed" + ) from e + except Exception: + validator_class.check_schema(schema) + raise + + class ConcurrentDeclarativeSource(Source): # By default, we defer to a value of 2. A value lower than could cause a PartitionEnqueuer to be stuck in a state of deadlock # because it has hit the limit of futures but not partition reader is consuming them. @@ -345,23 +366,7 @@ 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: - 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 error + _validate_manifest(self._source_config, self._declarative_component_schema) def _migrate_and_transform_config( self, diff --git a/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py b/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py index a7834f702f..c5dbbe8612 100644 --- a/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py +++ b/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py @@ -14,8 +14,8 @@ import pytest import requests import yaml -from jsonschema.exceptions import SchemaError, UnknownType, ValidationError -from jsonschema.validators import Draft7Validator, validate +from jsonschema.exceptions import ValidationError +from jsonschema.validators import validator_for import unit_tests.sources.declarative.external_component # Needed for dynamic imports to work from airbyte_cdk.legacy.sources.declarative.manifest_declarative_source import ( @@ -45,6 +45,9 @@ from airbyte_cdk.sources.declarative.retrievers.simple_retriever import SimpleRetriever from airbyte_cdk.sources.streams.concurrent.default_stream import DefaultStream from unit_tests.sources.declarative.parsers.test_model_to_component_factory import get_retriever +from unit_tests.sources.declarative.test_concurrent_declarative_source import ( + _two_stream_schema_validation_manifest, +) logger = logging.getLogger("airbyte") @@ -2554,103 +2557,26 @@ def capturing_resolve(resolver_self, *args, **kwargs): ) -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(): +@pytest.mark.parametrize("migrate_manifest", [False, True]) +def test_building_a_source_does_not_check_the_component_schema(migrate_manifest): with patch.object( - Draft7Validator, + validator_for(_get_declarative_component_schema()), "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" + ) as check_schema: + ManifestDeclarativeSource( + source_config=_two_stream_schema_validation_manifest(), + migrate_manifest=migrate_manifest, + ) + check_schema.assert_not_called() - assert list(Draft7Validator(broken).iter_errors(manifest)) +def test_validate_source_delegates_to_validate_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)) + "airbyte_cdk.legacy.sources.declarative.manifest_declarative_source._validate_manifest" + ) as validate_manifest: + source = ManifestDeclarativeSource(source_config=_two_stream_schema_validation_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) + validate_manifest.assert_called_once_with( + source._source_config, source._declarative_component_schema + ) diff --git a/unit_tests/sources/declarative/test_concurrent_declarative_source.py b/unit_tests/sources/declarative/test_concurrent_declarative_source.py index 7aa5775118..7549453e36 100644 --- a/unit_tests/sources/declarative/test_concurrent_declarative_source.py +++ b/unit_tests/sources/declarative/test_concurrent_declarative_source.py @@ -26,7 +26,7 @@ TraceType, ) from jsonschema.exceptions import SchemaError, UnknownType, ValidationError -from jsonschema.validators import Draft7Validator, validate +from jsonschema.validators import Draft7Validator, validate, validator_for from typing_extensions import deprecated import unit_tests.sources.declarative.external_component # Needed for dynamic imports to work @@ -7189,7 +7189,7 @@ def test_invalid_manifest_raises_the_best_match_error_not_the_first(): manifest["streams"][1]["retriever"]["requester"]["http_method"] = "FETCH" schema = _get_declarative_component_schema() - assert next(Draft7Validator(schema).iter_errors(manifest)).json_path == "$.streams[0]" + first = next(Draft7Validator(schema).iter_errors(manifest)) with pytest.raises(ValidationError) as expected_error: validate(manifest, schema) @@ -7206,32 +7206,36 @@ def test_invalid_manifest_raises_the_best_match_error_not_the_first(): ) cause = exc_info.value.__cause__ assert cause is not None + assert cause.__traceback__ 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" + assert cause.json_path != first.json_path -def test_building_a_source_does_not_check_the_component_schema(): +@pytest.mark.parametrize("migrate_manifest", [False, True]) +def test_building_a_source_does_not_check_the_component_schema(migrate_manifest): with patch.object( - Draft7Validator, + validator_for(_get_declarative_component_schema()), "check_schema", side_effect=AssertionError("check_schema must not run at source build"), - ): + ) as check_schema: ConcurrentDeclarativeSource( source_config=_two_stream_schema_validation_manifest(), config={}, catalog=create_catalog("a"), state=None, + migrate_manifest=migrate_manifest, ) + check_schema.assert_not_called() 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" + broken["definitions"]["HttpRequester"]["required"] = "type" assert list(Draft7Validator(broken).iter_errors(manifest)) @@ -7245,6 +7249,29 @@ def test_a_broken_component_schema_that_fails_the_manifest_raises_schema_error() ) +def test_a_component_schema_that_passes_check_schema_but_breaks_validation_reraises(): + manifest = _two_stream_schema_validation_manifest() + broken = copy.deepcopy(_get_declarative_component_schema()) + broken["definitions"]["HttpRequester"]["properties"]["path"] = { + "$ref": "#/definitions/DoesNotExist" + } + + Draft7Validator.check_schema(broken) + with pytest.raises(Exception) as expected: + 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(type(expected.value)) as exc_info: + ConcurrentDeclarativeSource( + source_config=manifest, config={}, catalog=create_catalog("a"), state=None + ) + + assert type(exc_info.value) is type(expected.value) + + 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()) diff --git a/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py index 3f8dee86b9..2a72934855 100644 --- a/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py +++ b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py @@ -12,11 +12,12 @@ at the wrong layer and with unreadable output. What these guards add is not coverage but a named failure in milliseconds. -Two guards, because the two ways prose breaks a plain scalar look nothing alike. A `": "` is loud: -the file stops parsing, so loading it is enough to catch it. A `" #"` is silent: YAML reads the -rest of the line as a comment, the value keeps parsing as a shorter string, and nothing in the +Three guards, because the two ways prose breaks a plain scalar look nothing alike. A `": "` is +loud: the file stops parsing, so loading it is enough to catch it. A `" #"` is silent: YAML reads +the rest of the line as a comment, the value keeps parsing as a shorter string, and nothing in the parsed tree records that a sentence went missing. Only the source text knows, so the second guard -reads that instead of the loaded schema. +reads that instead of the loaded schema. The third validates the parsed schema against Draft 7, +since `_validate_source` checks it only when a manifest fails validation. """ import pkgutil @@ -54,7 +55,7 @@ def test_the_shipped_component_schema_parses(): 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.""" + """`_validate_source` runs `check_schema` only when a manifest fails validation, so a malformed schema that still accepts manifests is caught only here.""" schema = _get_declarative_component_schema() validator_class = validator_for(schema) assert validator_class is Draft7Validator From bbff88e90768e4dd27c0ffbaed2766f1e5402395 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 21 Oct 2026 20:40:59 +0000 Subject: [PATCH 8/9] refactor(declarative): make validate_manifest public Co-Authored-By: bot_apk --- .../manifest_declarative_source.py | 4 +-- .../concurrent_declarative_source.py | 19 ++++++------ .../test_manifest_declarative_source.py | 14 ++++----- .../parsers/test_manifest_normalizer.py | 20 ++++++------- .../test_concurrent_declarative_source.py | 30 +++++++++---------- ...eclarative_component_schema_is_loadable.py | 8 ++--- 6 files changed, 48 insertions(+), 47 deletions(-) diff --git a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py index b8d6f5571b..1d06e25edc 100644 --- a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py +++ b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py @@ -34,7 +34,7 @@ from airbyte_cdk.models.airbyte_protocol_serializers import AirbyteMessageSerializer from airbyte_cdk.sources.declarative.checks import COMPONENTS_CHECKER_TYPE_MAPPING from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker -from airbyte_cdk.sources.declarative.concurrent_declarative_source import _validate_manifest +from airbyte_cdk.sources.declarative.concurrent_declarative_source import validate_manifest from airbyte_cdk.sources.declarative.interpolation import InterpolatedBoolean from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( ConditionalStreams as ConditionalStreamsModel, @@ -447,7 +447,7 @@ def _validate_source(self) -> None: Validates the connector manifest against the declarative component schema """ - _validate_manifest(self._source_config, self._declarative_component_schema) + validate_manifest(self._source_config, self._declarative_component_schema) cdk_version_str = metadata.version("airbyte_cdk") cdk_version = self._parse_version(cdk_version_str, "airbyte-cdk") diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index cc4688921b..e93437f02c 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -122,7 +122,7 @@ class TestLimits: max_streams: int = field(default=DEFAULT_MAX_STREAMS) -def _get_declarative_component_schema() -> Dict[str, Any]: +def get_declarative_component_schema() -> Dict[str, Any]: try: raw_component_schema = pkgutil.get_data( "airbyte_cdk", "sources/declarative/declarative_component_schema.yaml" @@ -140,12 +140,13 @@ def _get_declarative_component_schema() -> Dict[str, Any]: ) -def _validate_manifest(manifest: Mapping[str, Any], schema: Mapping[str, Any]) -> None: - # The component schema ships with the CDK and is checked against its metaschema in unit - # tests, so a valid manifest skips `check_schema`. If validation fails or raises, - # `check_schema` runs before the error propagates, so a broken schema surfaces as - # `SchemaError` instead of being blamed on the manifest. `best_match` keeps the error - # `jsonschema.validate` would raise; raising it here gives the cause a traceback. +def validate_manifest(manifest: Mapping[str, Any], schema: Mapping[str, Any]) -> None: + """Validate a manifest against the shipped component schema. + + Unit tests check the schema's metaschema, so valid manifests skip `check_schema`. On failure, + check the schema first to surface `SchemaError`, while preserving `jsonschema.validate`'s + best-match error as the cause with a traceback. + """ validator_class = validator_for(schema) try: error = best_match(validator_class(schema).iter_errors(manifest)) @@ -225,7 +226,7 @@ def __init__( self._should_normalize = normalize_manifest self._should_migrate = migrate_manifest - self._declarative_component_schema = _get_declarative_component_schema() + self._declarative_component_schema = get_declarative_component_schema() # If custom components are needed, locate and/or register them. self.components_module: ModuleType | None = get_registered_components_module(config=config) # set additional attributes @@ -366,7 +367,7 @@ def _validate_source(self) -> None: Validates the connector manifest against the declarative component schema """ - _validate_manifest(self._source_config, self._declarative_component_schema) + validate_manifest(self._source_config, self._declarative_component_schema) def _migrate_and_transform_config( self, diff --git a/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py b/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py index c5dbbe8612..b584f8b717 100644 --- a/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py +++ b/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py @@ -34,7 +34,7 @@ ) from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( ConcurrentDeclarativeSource, - _get_declarative_component_schema, + get_declarative_component_schema, ) from airbyte_cdk.sources.declarative.parsers.model_to_component_factory import ( ModelToComponentFactory, @@ -46,7 +46,7 @@ from airbyte_cdk.sources.streams.concurrent.default_stream import DefaultStream from unit_tests.sources.declarative.parsers.test_model_to_component_factory import get_retriever from unit_tests.sources.declarative.test_concurrent_declarative_source import ( - _two_stream_schema_validation_manifest, + two_stream_schema_validation_manifest, ) logger = logging.getLogger("airbyte") @@ -2560,22 +2560,22 @@ def capturing_resolve(resolver_self, *args, **kwargs): @pytest.mark.parametrize("migrate_manifest", [False, True]) def test_building_a_source_does_not_check_the_component_schema(migrate_manifest): with patch.object( - validator_for(_get_declarative_component_schema()), + validator_for(get_declarative_component_schema()), "check_schema", side_effect=AssertionError("check_schema must not run at source build"), ) as check_schema: ManifestDeclarativeSource( - source_config=_two_stream_schema_validation_manifest(), + source_config=two_stream_schema_validation_manifest(), migrate_manifest=migrate_manifest, ) check_schema.assert_not_called() -def test_validate_source_delegates_to_validate_manifest(): +def test_legacy_source_delegates_to_public_manifest_validator(): with patch( - "airbyte_cdk.legacy.sources.declarative.manifest_declarative_source._validate_manifest" + "airbyte_cdk.legacy.sources.declarative.manifest_declarative_source.validate_manifest" ) as validate_manifest: - source = ManifestDeclarativeSource(source_config=_two_stream_schema_validation_manifest()) + source = ManifestDeclarativeSource(source_config=two_stream_schema_validation_manifest()) validate_manifest.assert_called_once_with( source._source_config, source._declarative_component_schema diff --git a/unit_tests/sources/declarative/parsers/test_manifest_normalizer.py b/unit_tests/sources/declarative/parsers/test_manifest_normalizer.py index 3e934db792..246a840b1c 100644 --- a/unit_tests/sources/declarative/parsers/test_manifest_normalizer.py +++ b/unit_tests/sources/declarative/parsers/test_manifest_normalizer.py @@ -4,7 +4,7 @@ from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( - _get_declarative_component_schema, + get_declarative_component_schema, ) from airbyte_cdk.sources.declarative.parsers.manifest_normalizer import ( ManifestNormalizer, @@ -25,7 +25,7 @@ def test_when_multiple_url_base_are_resolved_and_most_frequent_is_shared( and the most frequent one is shared. """ - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() resolved_manifest = resolver.preprocess_manifest(manifest_with_multiple_url_base) normalized_manifest = ManifestNormalizer(resolved_manifest, schema).normalize() @@ -41,7 +41,7 @@ def test_with_shared_definitions_url_base_are_present( between the definitions and the `url_base` is present in the manifest. """ - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() resolved_manifest = resolver.preprocess_manifest(manifest_with_url_base_linked_definition) normalized_manifest = ManifestNormalizer(resolved_manifest, schema).normalize() @@ -75,7 +75,7 @@ def test_clean_dangling_fields_removes_unreferenced_definitions() -> None: } ], } - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() normalizer = ManifestNormalizer(manifest, schema) normalizer._clean_dangling_fields() assert normalizer._normalized_manifest == expected @@ -100,7 +100,7 @@ def test_clean_dangling_fields_removes_unreferenced_schemas() -> None: {"name": "stream1", "schema_loader": {"schema": {"$ref": "#/schemas/referenced"}}} ], } - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() normalizer = ManifestNormalizer(manifest, schema) normalizer._clean_dangling_fields() assert normalizer._normalized_manifest == expected @@ -134,7 +134,7 @@ def test_clean_dangling_fields_keeps_parent_paths() -> None: } ], } - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() normalizer = ManifestNormalizer(manifest, schema) normalizer._clean_dangling_fields() assert normalizer._normalized_manifest == expected @@ -150,7 +150,7 @@ def test_clean_dangling_fields_removes_empty_sections() -> None: "streams": [{"name": "stream1", "type": "object", "properties": {}}], } expected = {"streams": [{"name": "stream1", "type": "object", "properties": {}}]} - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() normalizer = ManifestNormalizer(manifest, schema) normalizer._clean_dangling_fields() assert normalizer._normalized_manifest == expected @@ -220,7 +220,7 @@ def test_replace_parent_streams_with_refs_replaces_with_ref(): }, ] } - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() normalizer = ManifestNormalizer(manifest, schema) normalizer._replace_parent_streams_with_refs() assert normalizer._normalized_manifest == expected @@ -276,7 +276,7 @@ def test_replace_parent_streams_with_refs_no_match(): }, ] } - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() normalizer = ManifestNormalizer(manifest, schema) normalizer._replace_parent_streams_with_refs() assert normalizer._normalized_manifest == expected @@ -360,7 +360,7 @@ def test_replace_parent_streams_with_refs_handles_multiple_partition_routers(): }, ] } - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() normalizer = ManifestNormalizer(manifest, schema) normalizer._replace_parent_streams_with_refs() assert normalizer._normalized_manifest == expected diff --git a/unit_tests/sources/declarative/test_concurrent_declarative_source.py b/unit_tests/sources/declarative/test_concurrent_declarative_source.py index 7549453e36..dc82ca4d2b 100644 --- a/unit_tests/sources/declarative/test_concurrent_declarative_source.py +++ b/unit_tests/sources/declarative/test_concurrent_declarative_source.py @@ -55,7 +55,7 @@ from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( ConcurrentDeclarativeSource, TestLimits, - _get_declarative_component_schema, + get_declarative_component_schema, ) from airbyte_cdk.sources.declarative.extractors.record_filter import ( ClientSideIncrementalRecordFilterDecorator, @@ -7150,7 +7150,7 @@ def test_combined_extractor_skipping_empty_records_reads_every_page( ] -def _two_stream_schema_validation_manifest() -> Dict[str, Any]: +def two_stream_schema_validation_manifest() -> Dict[str, Any]: def stream(name: str) -> Dict[str, Any]: return { "type": "DeclarativeStream", @@ -7184,10 +7184,10 @@ def stream(name: str) -> Dict[str, Any]: def test_invalid_manifest_raises_the_best_match_error_not_the_first(): - manifest = _two_stream_schema_validation_manifest() + 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() + schema = get_declarative_component_schema() first = next(Draft7Validator(schema).iter_errors(manifest)) @@ -7218,12 +7218,12 @@ def test_invalid_manifest_raises_the_best_match_error_not_the_first(): @pytest.mark.parametrize("migrate_manifest", [False, True]) def test_building_a_source_does_not_check_the_component_schema(migrate_manifest): with patch.object( - validator_for(_get_declarative_component_schema()), + validator_for(get_declarative_component_schema()), "check_schema", side_effect=AssertionError("check_schema must not run at source build"), ) as check_schema: ConcurrentDeclarativeSource( - source_config=_two_stream_schema_validation_manifest(), + source_config=two_stream_schema_validation_manifest(), config={}, catalog=create_catalog("a"), state=None, @@ -7233,14 +7233,14 @@ def test_building_a_source_does_not_check_the_component_schema(migrate_manifest) 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()) + manifest = two_stream_schema_validation_manifest() + broken = copy.deepcopy(get_declarative_component_schema()) broken["definitions"]["HttpRequester"]["required"] = "type" assert list(Draft7Validator(broken).iter_errors(manifest)) with patch( - "airbyte_cdk.sources.declarative.concurrent_declarative_source._get_declarative_component_schema", + "airbyte_cdk.sources.declarative.concurrent_declarative_source.get_declarative_component_schema", return_value=broken, ): with pytest.raises(SchemaError): @@ -7250,8 +7250,8 @@ def test_a_broken_component_schema_that_fails_the_manifest_raises_schema_error() def test_a_component_schema_that_passes_check_schema_but_breaks_validation_reraises(): - manifest = _two_stream_schema_validation_manifest() - broken = copy.deepcopy(_get_declarative_component_schema()) + manifest = two_stream_schema_validation_manifest() + broken = copy.deepcopy(get_declarative_component_schema()) broken["definitions"]["HttpRequester"]["properties"]["path"] = { "$ref": "#/definitions/DoesNotExist" } @@ -7261,7 +7261,7 @@ def test_a_component_schema_that_passes_check_schema_but_breaks_validation_rerai list(Draft7Validator(broken).iter_errors(manifest)) with patch( - "airbyte_cdk.sources.declarative.concurrent_declarative_source._get_declarative_component_schema", + "airbyte_cdk.sources.declarative.concurrent_declarative_source.get_declarative_component_schema", return_value=broken, ): with pytest.raises(type(expected.value)) as exc_info: @@ -7273,15 +7273,15 @@ def test_a_component_schema_that_passes_check_schema_but_breaks_validation_rerai 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()) + 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", + "airbyte_cdk.sources.declarative.concurrent_declarative_source.get_declarative_component_schema", return_value=broken, ): with pytest.raises(SchemaError): diff --git a/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py index 2a72934855..bdab6a30d8 100644 --- a/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py +++ b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py @@ -4,7 +4,7 @@ """The shipped manifest schema has to parse, and only 194 unrelated failures said so. -`_get_declarative_component_schema()` runs before any low-code source is built, so a malformed +`get_declarative_component_schema()` runs before any low-code source is built, so a malformed `declarative_component_schema.yaml` fails every declarative connector at startup. It is also easy to break from a documentation edit alone: the descriptions are plain YAML scalars, so a `": "` in prose ends the scalar and the file stops being a mapping. That happened in this file's history and @@ -26,7 +26,7 @@ from jsonschema.validators import Draft7Validator, validator_for from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( - _get_declarative_component_schema, + get_declarative_component_schema, ) # A plain scalar is one not opened with a quote, a block indicator or a flow collection, and it is @@ -48,7 +48,7 @@ def test_the_shipped_component_schema_parses(): - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() assert schema["title"] == "DeclarativeSource" assert "HttpRequester" in schema["definitions"] @@ -56,7 +56,7 @@ def test_the_shipped_component_schema_parses(): def test_the_shipped_component_schema_is_a_valid_draft_7_schema(): """`_validate_source` runs `check_schema` only when a manifest fails validation, so a malformed schema that still accepts manifests is caught only here.""" - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() validator_class = validator_for(schema) assert validator_class is Draft7Validator validator_class.check_schema(schema) From 6574c476341f128705df86af01e7efc7f316a23b Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 21 Oct 2026 21:51:32 +0000 Subject: [PATCH 9/9] fix(sdm): warn on compile failures, reject unvalidatable .pyc, skip test dirs Co-Authored-By: bot_apk --- .github/workflows/docker-build-check.yml | 9 +++++++-- Dockerfile | 9 +++++---- 2 files changed, 12 insertions(+), 6 deletions(-) diff --git a/.github/workflows/docker-build-check.yml b/.github/workflows/docker-build-check.yml index f20e6e8b85..52c90a2353 100644 --- a/.github/workflows/docker-build-check.yml +++ b/.github/workflows/docker-build-check.yml @@ -58,14 +58,19 @@ jobs: docker run --rm -i --entrypoint python airbyte/source-declarative-manifest:bytecode-check - <<'EOF' import importlib.util import pathlib + import re import struct import sys import sysconfig + SKIP = re.compile(r"/tests?/") # keep in sync with compileall -x in the Dockerfile + root = pathlib.Path(sysconfig.get_paths()["purelib"]) missing, stale = [], [] total = 0 for src in root.rglob("*.py"): + if SKIP.search(str(src)): + continue total += 1 pyc = pathlib.Path(importlib.util.cache_from_source(str(src))) if not pyc.is_file(): @@ -73,9 +78,9 @@ jobs: continue header = pyc.read_bytes()[:16] flags = int.from_bytes(header[4:8], "little") - if header[:4] != importlib.util.MAGIC_NUMBER: + if len(header) < 16 or header[:4] != importlib.util.MAGIC_NUMBER or flags != 0: stale.append(src) - elif flags == 0: + else: mtime, size = struct.unpack("