|
| 1 | +# |
| 2 | +# Copyright (c) 2024 Airbyte, Inc., all rights reserved. |
| 3 | +# |
| 4 | + |
| 5 | +from copy import deepcopy |
| 6 | +from dataclasses import InitVar, dataclass, field |
| 7 | +from typing import Any, Dict, Iterable, List, Mapping |
| 8 | + |
| 9 | +import dpath |
| 10 | +import yaml |
| 11 | +from typing_extensions import deprecated |
| 12 | +from yaml.parser import ParserError |
| 13 | + |
| 14 | +from airbyte_cdk.sources.declarative.interpolation import InterpolatedString |
| 15 | +from airbyte_cdk.sources.declarative.resolvers.components_resolver import ( |
| 16 | + ComponentMappingDefinition, |
| 17 | + ComponentsResolver, |
| 18 | + ResolvedComponentMappingDefinition, |
| 19 | +) |
| 20 | +from airbyte_cdk.sources.source import ExperimentalClassWarning |
| 21 | +from airbyte_cdk.sources.types import Config |
| 22 | + |
| 23 | + |
| 24 | +@deprecated("This class is experimental. Use at your own risk.", category=ExperimentalClassWarning) |
| 25 | +@dataclass |
| 26 | +class StreamParametersDefinition: |
| 27 | + """ |
| 28 | + Represents a stream parameters definition to set up dynamic streams from defined values in manifest. |
| 29 | + """ |
| 30 | + |
| 31 | + list_of_parameters_for_stream: List[Dict[str, Any]] |
| 32 | + |
| 33 | + |
| 34 | +@deprecated("This class is experimental. Use at your own risk.", category=ExperimentalClassWarning) |
| 35 | +@dataclass |
| 36 | +class ParametrizedComponentsResolver(ComponentsResolver): |
| 37 | + """ |
| 38 | + Resolves and populates dynamic streams from defined parametrized values in manifest. |
| 39 | + """ |
| 40 | + |
| 41 | + stream_parameters: StreamParametersDefinition |
| 42 | + config: Config |
| 43 | + components_mapping: List[ComponentMappingDefinition] |
| 44 | + parameters: InitVar[Mapping[str, Any]] |
| 45 | + _resolved_components: List[ResolvedComponentMappingDefinition] = field( |
| 46 | + init=False, repr=False, default_factory=list |
| 47 | + ) |
| 48 | + |
| 49 | + def __post_init__(self, parameters: Mapping[str, Any]) -> None: |
| 50 | + """ |
| 51 | + Initializes and parses component mappings, converting them to resolved definitions. |
| 52 | +
|
| 53 | + Args: |
| 54 | + parameters (Mapping[str, Any]): Parameters for interpolation. |
| 55 | + """ |
| 56 | + |
| 57 | + for component_mapping in self.components_mapping: |
| 58 | + if isinstance(component_mapping.value, (str, InterpolatedString)): |
| 59 | + interpolated_value = ( |
| 60 | + InterpolatedString.create(component_mapping.value, parameters=parameters) |
| 61 | + if isinstance(component_mapping.value, str) |
| 62 | + else component_mapping.value |
| 63 | + ) |
| 64 | + |
| 65 | + field_path = [ |
| 66 | + InterpolatedString.create(path, parameters=parameters) |
| 67 | + for path in component_mapping.field_path |
| 68 | + ] |
| 69 | + |
| 70 | + self._resolved_components.append( |
| 71 | + ResolvedComponentMappingDefinition( |
| 72 | + field_path=field_path, |
| 73 | + value=interpolated_value, |
| 74 | + value_type=component_mapping.value_type, |
| 75 | + create_or_update=component_mapping.create_or_update, |
| 76 | + parameters=parameters, |
| 77 | + ) |
| 78 | + ) |
| 79 | + else: |
| 80 | + raise ValueError( |
| 81 | + f"Expected a string or InterpolatedString for value in mapping: {component_mapping}" |
| 82 | + ) |
| 83 | + |
| 84 | + def resolve_components( |
| 85 | + self, stream_template_config: Dict[str, Any] |
| 86 | + ) -> Iterable[Dict[str, Any]]: |
| 87 | + kwargs = {"stream_template_config": stream_template_config} |
| 88 | + |
| 89 | + for components_values in self.stream_parameters.list_of_parameters_for_stream: |
| 90 | + updated_config = deepcopy(stream_template_config) |
| 91 | + kwargs["components_values"] = components_values # type: ignore[assignment] # component_values will always be of type Mapping[str, Any] |
| 92 | + for resolved_component in self._resolved_components: |
| 93 | + valid_types = ( |
| 94 | + (resolved_component.value_type,) if resolved_component.value_type else None |
| 95 | + ) |
| 96 | + value = resolved_component.value.eval( |
| 97 | + self.config, valid_types=valid_types, **kwargs |
| 98 | + ) |
| 99 | + path = [path.eval(self.config, **kwargs) for path in resolved_component.field_path] |
| 100 | + parsed_value = self._parse_yaml_if_possible(value) |
| 101 | + # https://github.com/dpath-maintainers/dpath-python/blob/master/dpath/__init__.py#L136 |
| 102 | + # dpath.set returns the number of changed elements, 0 when no elements changed |
| 103 | + updated = dpath.set(updated_config, path, parsed_value) |
| 104 | + |
| 105 | + if parsed_value and not updated and resolved_component.create_or_update: |
| 106 | + dpath.new(updated_config, path, parsed_value) |
| 107 | + |
| 108 | + yield updated_config |
| 109 | + |
| 110 | + @staticmethod |
| 111 | + def _parse_yaml_if_possible(value: Any) -> Any: |
| 112 | + """ |
| 113 | + Try to turn value into a Python object by YAML-parsing it. |
| 114 | +
|
| 115 | + * If value is a `str` and can be parsed by `yaml.safe_load`, |
| 116 | + return the parsed result. |
| 117 | + * If parsing fails (`yaml.parser.ParserError`) – or value is not |
| 118 | + a string at all – return the original value unchanged. |
| 119 | + """ |
| 120 | + if isinstance(value, str): |
| 121 | + try: |
| 122 | + return yaml.safe_load(value) |
| 123 | + except ParserError: # "{{ record[0] in ['cohortActiveUsers'] }}" # not valid YAML |
| 124 | + return value |
| 125 | + return value |
0 commit comments