diff --git a/airbyte_cdk/sources/utils/schema_helpers.py b/airbyte_cdk/sources/utils/schema_helpers.py index ab7fbbf459..9c672edd4b 100644 --- a/airbyte_cdk/sources/utils/schema_helpers.py +++ b/airbyte_cdk/sources/utils/schema_helpers.py @@ -11,7 +11,7 @@ from typing import TYPE_CHECKING, Any, ClassVar, Dict, List, Mapping, MutableMapping, Tuple, cast import jsonref -from jsonschema import validate +from jsonschema import Draft7Validator from jsonschema.exceptions import ValidationError from pydantic.v1 import BaseModel, Field from referencing import Registry, Resource @@ -185,25 +185,66 @@ def _resolve_schema_references(self, raw_schema: dict[str, Any]) -> dict[str, An raise ValueError(f"Expected resolved to be a dict. Got {resolved}") +def _format_oneof_error(error: ValidationError) -> str: + """Format a oneOf/anyOf validation error with per-branch details.""" + field_path = ".".join(str(p) for p in error.absolute_path) if error.absolute_path else "root" + branch_schemas = error.schema.get(error.validator, []) + branch_messages: list[str] = [] + errors_by_branch: dict[int, list[str]] = {} + for sub_error in error.context: + branch_index = sub_error.schema_path[0] + if isinstance(branch_index, int): + errors_by_branch.setdefault(branch_index, []).append(sub_error.message) + + for i, branch_schema in enumerate(branch_schemas): + label = branch_schema.get("title", f"option {i + 1}") + sub_errors = errors_by_branch.get(i, []) + if sub_errors: + details = "; ".join(sub_errors) + branch_messages.append(f"{label}: {details}") + else: + branch_messages.append(f"{label}: unknown validation failure") + + joined = " | ".join(branch_messages) + return f'Config validation error: the provided configuration for "{field_path}" does not match any supported option. {joined}.' + + def check_config_against_spec_or_exit( config: Mapping[str, Any], spec: ConnectorSpecification ) -> None: - """ - Check config object against spec. In case of spec is invalid, throws - an exception with validation error description. + """Check config object against spec and raise on validation failure. - :param config - config loaded from file specified over command line - :param spec - spec object generated by connector + Uses iter_errors to detect oneOf/anyOf failures and produce a message + that lists per-branch validation details instead of surfacing a + misleading error from the last non-matching branch. """ spec_schema = spec.connectionSpecification - try: - validate(instance=config, schema=spec_schema) - except ValidationError as validation_error: - raise AirbyteTracedException( - message="Config validation error: " + validation_error.message, - internal_message=validation_error.message, - failure_type=FailureType.config_error, - ) from None # required to prevent logging config secrets from the ValidationError's stacktrace + validator = Draft7Validator(spec_schema) + errors = list(validator.iter_errors(config)) + if not errors: + return + + oneof_errors = [e for e in errors if e.validator in ("oneOf", "anyOf") and e.context] + if oneof_errors: + error = oneof_errors[0] + message = _format_oneof_error(error) + internal_parts = [ + f"{sub.message} (path: {list(sub.absolute_path)})" for sub in error.context + ] + internal_message = ( + f"oneOf validation failed at path {list(error.absolute_path)}: " + + "; ".join(internal_parts) + ) + else: + error = errors[0] + message = "Config validation error: " + error.message + internal_message = error.message + + raise AirbyteTracedException( + message=message, + internal_message=internal_message, + failure_type=FailureType.config_error, + ) from None # required to prevent logging config secrets from the ValidationError's stacktrace class InternalConfig(BaseModel):