airbyte_cdk.sources.declarative.validators

10class Validator(ABC):
11    @abstractmethod
12    def validate(self, input_data: Any) -> None:
13        """
14        Validates the input data.
15
16        :param input_data: The data to validate
17        :raises ValueError: If validation fails
18        """
19        pass

19@dataclass
20class DpathValidator(Validator):
21    """
22    Validator that extracts a value at a specific path in the input data
23    and applies a validation strategy to it.
24    """
25
26    field_path: List[str]
27    strategy: ValidationStrategy
28
29    def __post_init__(self) -> None:
30        self._field_path = [
31            InterpolatedString.create(path, parameters={}) for path in self.field_path
32        ]
33        for path_index in range(len(self.field_path)):
34            if isinstance(self.field_path[path_index], str):
35                self._field_path[path_index] = InterpolatedString.create(
36                    self.field_path[path_index], parameters={}
37                )
38
39    def validate(self, input_data: dict[str, Any]) -> None:
40        """
41        Extracts the value at the specified path and applies the validation strategy.
42
43        :param input_data: Dictionary containing the data to validate
44        :raises ValueError: If the path doesn't exist or validation fails
45        """
46        path = [path.eval({}) for path in self._field_path]
47
48        if len(path) == 0:
49            raise ValueError("Field path is empty")
50
51        if "*" in path:
52            try:
53                values = dpath.values(input_data, path)
54                for value in values:
55                    self.strategy.validate(value)
56            except KeyError as e:
57                logger.warning(f"Error validating path. Key not found: {e}")
58                return
59
60        else:
61            try:
62                value = dpath.get(input_data, path)
63                self.strategy.validate(value)
64            except KeyError as e:
65                logger.warning(f"Error validating path. Key not found: {e}")
66                return

10class ValidationStrategy(ABC):
11    """
12    Base class for validation strategies.
13    """
14
15    @abstractmethod
16    def validate(self, value: Any) -> None:
17        """
18        Validates a value according to a specific strategy.
19
20        :param value: The value to validate
21        :raises ValueError: If validation fails
22        """
23        pass

15@dataclass
16class ValidateAdheresToSchema(ValidationStrategy):
17    """
18    Validates that a value adheres to a specified JSON schema.
19    """
20
21    schema: Mapping[str, Any]
22
23    def validate(self, value: Any) -> None:
24        """
25        Validates the value against the JSON schema.
26
27        :param value: The value to validate
28        :raises ValueError: If the value does not adhere to the schema
29        """
30
31        if isinstance(value, str):
32            try:
33                value = json.loads(value)
34            except json.JSONDecodeError as e:
35                raise ValueError(f"Invalid JSON string: {value}") from e
36
37        try:
38            jsonschema.validate(instance=value, schema=self.schema)
39        except jsonschema.ValidationError as e:
40            raise ValueError(f"JSON schema validation error: {e.message}")

12@dataclass
13class PredicateValidator:
14    """
15    Validator that applies a validation strategy to a value.
16    """
17
18    value: Any
19    strategy: ValidationStrategy
20
21    def validate(self) -> None:
22        """
23        Applies the validation strategy to the value.
24
25        :raises ValueError: If validation fails
26        """
27        self.strategy.validate(self.value)