airbyte_cdk.sources.declarative.interpolation

29@dataclass
30class InterpolatedBoolean:
31    f"""
32    Wrapper around a string to be evaluated to a boolean value.
33    The string will be evaluated as False if it interpolates to a value in {FALSE_VALUES}
34
35    Attributes:
36        condition (str): The string representing the condition to evaluate to a boolean
37    """
38    condition: str
39    parameters: InitVar[Mapping[str, Any]]
40
41    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
42        self._default = "False"
43        self._interpolation = JinjaInterpolation()
44        self._parameters = parameters
45
46    def eval(self, config: Config, **additional_parameters: Any) -> bool:
47        """
48        Interpolates the predicate condition string using the config and other optional arguments passed as parameter.
49
50        :param config: The user-provided configuration as specified by the source's spec
51        :param additional_parameters: Optional parameters used for interpolation
52        :return: The interpolated string
53        """
54        if isinstance(self.condition, bool):
55            return self.condition
56        else:
57            evaluated = self._interpolation.eval(
58                self.condition,
59                config,
60                self._default,
61                parameters=self._parameters,
62                **additional_parameters,
63            )
64            if evaluated in FALSE_VALUES:
65                return False
66            # The presence of a value is generally regarded as truthy, so we treat it as such
67            return True

14@dataclass
15class InterpolatedMapping:
16    """
17    Wrapper around a Mapping[str, str] where both the keys and values are to be interpolated.
18
19    Attributes:
20        mapping (Mapping[str, str]): to be evaluated
21    """
22
23    mapping: Mapping[str, str]
24    parameters: InitVar[Mapping[str, Any]]
25
26    def __post_init__(self, parameters: Optional[Mapping[str, Any]]) -> None:
27        self._interpolation = JinjaInterpolation()
28        self._parameters = parameters
29
30    def eval(self, config: Config, **additional_parameters: Any) -> Dict[str, Any]:
31        """
32        Wrapper around a Mapping[str, str] that allows for both keys and values to be interpolated.
33
34        :param config: The user-provided configuration as specified by the source's spec
35        :param additional_parameters: Optional parameters used for interpolation
36        :return: The interpolated mapping
37        """
38        valid_key_types = additional_parameters.pop("valid_key_types", (str,))
39        valid_value_types = additional_parameters.pop("valid_value_types", None)
40        return {
41            self._interpolation.eval(
42                name,
43                config,
44                valid_types=valid_key_types,
45                parameters=self._parameters,
46                **additional_parameters,
47            ): self._eval(value, config, valid_types=valid_value_types, **additional_parameters)
48            for name, value in self.mapping.items()
49        }
50
51    def _eval(self, value: str, config: Config, **kwargs: Any) -> Any:
52        # The values in self._mapping can be of Any type
53        # We only want to interpolate them if they are strings
54        if isinstance(value, str):
55            return self._interpolation.eval(value, config, parameters=self._parameters, **kwargs)
56        else:
57            return value

13@dataclass
14class InterpolatedString:
15    """
16    Wrapper around a raw string to be interpolated with the Jinja2 templating engine
17
18    Attributes:
19        string (str): The string to evalute
20        default (Optional[str]): The default value to return if the evaluation returns an empty string
21        parameters (Mapping[str, Any]): Additional runtime parameters to be used for string interpolation
22    """
23
24    string: str
25    parameters: InitVar[Mapping[str, Any]]
26    default: Optional[str] = None
27
28    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
29        self.default = self.default or self.string
30        self._interpolation = JinjaInterpolation()
31        self._parameters = parameters
32        # indicates whether passed string is just a plain string, not Jinja template
33        # This allows for optimization, but we do not know it yet at this stage
34        self._is_plain_string = None
35
36    def eval(self, config: Config, **kwargs: Any) -> Any:
37        """
38        Interpolates the input string using the config and other optional arguments passed as parameter.
39
40        :param config: The user-provided configuration as specified by the source's spec
41        :param kwargs: Optional parameters used for interpolation
42        :return: The interpolated string
43        """
44        if self._is_plain_string:
45            return self.string
46        if self._is_plain_string is None:
47            # Let's check whether output from evaluation is the same as input.
48            # This indicates occurrence of a plain string, not a template and we can skip Jinja in subsequent runs.
49            evaluated = self._interpolation.eval(
50                self.string, config, self.default, parameters=self._parameters, **kwargs
51            )
52            self._is_plain_string = self.string == evaluated
53            return evaluated
54        return self._interpolation.eval(
55            self.string, config, self.default, parameters=self._parameters, **kwargs
56        )
57
58    def __eq__(self, other: Any) -> bool:
59        if not isinstance(other, InterpolatedString):
60            return False
61        return self.string == other.string and self.default == other.default
62
63    @classmethod
64    def create(
65        cls,
66        string_or_interpolated: Union["InterpolatedString", str],
67        *,
68        parameters: Mapping[str, Any],
69    ) -> "InterpolatedString":
70        """
71        Helper function to obtain an InterpolatedString from either a raw string or an InterpolatedString.
72
73        :param string_or_interpolated: either a raw string or an InterpolatedString.
74        :param parameters: parameters propagated from parent component
75        :return: InterpolatedString representing the input string.
76        """
77        if isinstance(string_or_interpolated, str):
78            return InterpolatedString(string=string_or_interpolated, parameters=parameters)
79        else:
80            return string_or_interpolated