implement validators and transformer

Author: pnilan

airbyte_cdk/sources/declarative/transformations/__init__.py

@@ -12,6 +12,7 @@
 
 # isort: split
 from .add_fields import AddFields
+from .remap_field import RemapField
 from .remove_fields import RemoveFields
 
-__all__ = ["AddFields", "RecordTransformation", "RemoveFields"]
+__all__ = ["AddFields", "RecordTransformation", "RemapField", "RemoveFields"]

airbyte_cdk/sources/declarative/transformations/remap_field.py

@@ -0,0 +1,58 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import dataclass
+from typing import Any, Dict, Mapping, Optional
+
+from airbyte_cdk.sources.declarative.transformations.transformation import RecordTransformation
+from airbyte_cdk.sources.types import Config, StreamSlice, StreamState
+
+
+@dataclass
+class RemapField(RecordTransformation):
+    """
+    Transformation that remaps a field's value to another value based on a static map.
+    """
+
+    map: Mapping[str, Any]
+    field_path: str
+
+    def transform(
+        self,
+        record: Dict[str, Any],
+        config: Optional[Config] = None,
+        stream_state: Optional[StreamState] = None,
+        stream_slice: Optional[StreamSlice] = None,
+    ) -> None:
+        """
+        Transforms a record by remapping a field value based on the provided map.
+        If the original value is found in the map, it's replaced with the mapped value.
+        If the value is not in the map, the field remains unchanged.
+
+        :param record: The input record to be transformed
+        :param config: The user-provided configuration as specified by the source's spec
+        :param stream_state: The stream state
+        :param stream_slice: The stream slice
+        """
+        # Extract path components
+        path_components = self.field_path.split(".")
+
+        # Navigate to the parent object containing the field to remap
+        current = record
+        for i, component in enumerate(path_components[:-1]):
+            if component not in current:
+                # Path doesn't exist, so nothing to remap
+                return
+            current = current[component]
+
+            # If we encounter a non-dict, we can't continue navigating
+            if not isinstance(current, dict):
+                return
+
+        # The last component is the field name to remap
+        field_name = path_components[-1]
+
+        # Check if the field exists and remap its value if it's in the map
+        if field_name in current and current[field_name] in self.map:
+            current[field_name] = self.map[current[field_name]]

airbyte_cdk/sources/declarative/validators/__init__.py

@@ -0,0 +1,23 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from airbyte_cdk.sources.declarative.validators.dpath_validator import DpathValidator
+from airbyte_cdk.sources.declarative.validators.predicate_validator import PredicateValidator
+from airbyte_cdk.sources.declarative.validators.validate_adheres_to_schema import (
+    ValidateAdheresToSchema,
+)
+from airbyte_cdk.sources.declarative.validators.validate_is_in_list import ValidateIsInList
+from airbyte_cdk.sources.declarative.validators.validate_is_of_type import ValidateIsOfType
+from airbyte_cdk.sources.declarative.validators.validation_strategy import ValidationStrategy
+from airbyte_cdk.sources.declarative.validators.validator import Validator
+
+__all__ = [
+    "Validator",
+    "DpathValidator",
+    "ValidationStrategy",
+    "ValidateIsInList",
+    "ValidateIsOfType",
+    "ValidateAdheresToSchema",
+    "PredicateValidator",
+]

airbyte_cdk/sources/declarative/validators/dpath_validator.py

@@ -0,0 +1,49 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import dataclass
+from typing import Any, List, Union
+
+import dpath.util
+
+from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
+from airbyte_cdk.sources.declarative.validators.validator import Validator
+from airbyte_cdk.sources.declarative.validators.validation_strategy import ValidationStrategy
+
+
+@dataclass
+class DpathValidator(Validator):
+    """
+    Validator that extracts a value at a specific path in the input data
+    and applies a validation strategy to it.
+    """
+
+    field_path: List[Union[InterpolatedString, str]]
+    strategy: ValidationStrategy
+
+    def __post_init__(self) -> None:
+        self._field_path = [
+            InterpolatedString.create(path, parameters={}) for path in self.field_path
+        ]
+        for path_index in range(len(self.field_path)):
+            if isinstance(self.field_path[path_index], str):
+                self._field_path[path_index] = InterpolatedString.create(
+                    self.field_path[path_index], parameters={}
+                )
+
+    def validate(self, input_data: dict[str, Any]) -> None:
+        """
+        Extracts the value at the specified path and applies the validation strategy.
+
+        :param input_data: Dictionary containing the data to validate
+        :raises ValueError: If the path doesn't exist or validation fails
+        """
+        try:
+            path = [path.eval(input_data) for path in self._field_path]
+            value = dpath.util.get(input_data, path)
+            self.strategy.validate(value)
+        except KeyError:
+            raise ValueError(f"Path '{self.field_path}' not found in the input data")
+        except Exception as e:
+            raise ValueError(f"Error validating path '{self.field_path}': {e}")

airbyte_cdk/sources/declarative/validators/predicate_validator.py

@@ -0,0 +1,26 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import dataclass
+from typing import Any
+
+from airbyte_cdk.sources.declarative.validators.validation_strategy import ValidationStrategy
+
+
+@dataclass
+class PredicateValidator:
+    """
+    Validator that applies a validation strategy to a value.
+    """
+
+    value: Any
+    strategy: ValidationStrategy
+
+    def validate(self) -> None:
+        """
+        Applies the validation strategy to the value.
+
+        :raises ValueError: If validation fails
+        """
+        self.strategy.validate(self.value)

airbyte_cdk/sources/declarative/validators/validate_adheres_to_schema.py

@@ -0,0 +1,31 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import dataclass
+from typing import Any
+
+import jsonschema
+
+from airbyte_cdk.sources.declarative.validators.validation_strategy import ValidationStrategy
+
+
+@dataclass
+class ValidateAdheresToSchema(ValidationStrategy):
+    """
+    Validates that a value adheres to a specified JSON schema.
+    """
+
+    schema: dict[str, Any]
+
+    def validate(self, value: Any) -> None:
+        """
+        Validates the value against the JSON schema.
+
+        :param value: The value to validate
+        :raises ValueError: If the value does not adhere to the schema
+        """
+        try:
+            jsonschema.validate(instance=value, schema=self.schema)
+        except jsonschema.ValidationError as e:
+            raise ValueError(f"JSON schema validation error: {e.message}")

airbyte_cdk/sources/declarative/validators/validate_is_in_list.py

@@ -0,0 +1,28 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import dataclass
+from typing import Any
+
+from airbyte_cdk.sources.declarative.validators.validation_strategy import ValidationStrategy
+
+
+@dataclass
+class ValidateIsInList(ValidationStrategy):
+    """
+    Validates that a value is in a list of supported values.
+    """
+
+    supported_values: list[Any]
+
+    def validate(self, value: Any) -> None:
+        """
+        Checks if the value is in the list of supported values.
+
+        :param value: The value to validate
+        :raises ValueError: If the value is not in the list of supported values
+        """
+        if value not in self.supported_values:
+            supported_values_str = ", ".join(str(v) for v in self.supported_values)
+            raise ValueError(f"Value '{value}' not in supported values: [{supported_values_str}]")

airbyte_cdk/sources/declarative/validators/validate_is_of_type.py

@@ -0,0 +1,27 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import dataclass
+from typing import Any
+
+from airbyte_cdk.sources.declarative.validators.validation_strategy import ValidationStrategy
+
+
+@dataclass
+class ValidateIsOfType(ValidationStrategy):
+    """
+    Validates that a value is of a specified type.
+    """
+
+    expected_type: Any
+
+    def validate(self, value: Any) -> None:
+        """
+        Checks if the value is of the expected type.
+
+        :param value: The value to validate
+        :raises ValueError: If the value is not of the expected type
+        """
+        if not isinstance(value, self.expected_type):
+            raise ValueError(f"Value '{value}' is not of type {self.expected_type.__name__}")

airbyte_cdk/sources/declarative/validators/validation_strategy.py

@@ -0,0 +1,22 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class ValidationStrategy(ABC):
+    """
+    Base class for validation strategies.
+    """
+
+    @abstractmethod
+    def validate(self, value: Any) -> None:
+        """
+        Validates a value according to a specific strategy.
+
+        :param value: The value to validate
+        :raises ValueError: If validation fails
+        """
+        pass

airbyte_cdk/sources/declarative/validators/validator.py

@@ -0,0 +1,18 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class Validator(ABC):
+    @abstractmethod
+    def validate(self, input_data: Any) -> None:
+        """
+        Validates the input data.
+
+        :param input_data: The data to validate
+        :raises ValueError: If validation fails
+        """
+        pass