-
Notifications
You must be signed in to change notification settings - Fork 8
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(low-code): add DpathFlattenFields (#227)
Co-authored-by: octavia-squidington-iii <[email protected]>
- Loading branch information
1 parent
34a978d
commit 17dd71f
Showing
5 changed files
with
235 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
55 changes: 55 additions & 0 deletions
55
airbyte_cdk/sources/declarative/transformations/dpath_flatten_fields.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| from dataclasses import InitVar, dataclass | ||
| from typing import Any, Dict, List, Mapping, Optional, Union | ||
|
|
||
| import dpath | ||
|
|
||
| from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString | ||
| from airbyte_cdk.sources.declarative.transformations import RecordTransformation | ||
| from airbyte_cdk.sources.types import Config, StreamSlice, StreamState | ||
|
|
||
|
|
||
| @dataclass | ||
| class DpathFlattenFields(RecordTransformation): | ||
| """ | ||
| Flatten fields only for provided path. | ||
| field_path: List[Union[InterpolatedString, str]] path to the field to flatten. | ||
| delete_origin_value: bool = False whether to delete origin field or keep it. Default is False. | ||
| """ | ||
|
|
||
| config: Config | ||
| field_path: List[Union[InterpolatedString, str]] | ||
| parameters: InitVar[Mapping[str, Any]] | ||
| delete_origin_value: bool = False | ||
|
|
||
| def __post_init__(self, parameters: Mapping[str, Any]) -> None: | ||
| self._field_path = [ | ||
| InterpolatedString.create(path, parameters=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=parameters | ||
| ) | ||
|
|
||
| def transform( | ||
| self, | ||
| record: Dict[str, Any], | ||
| config: Optional[Config] = None, | ||
| stream_state: Optional[StreamState] = None, | ||
| stream_slice: Optional[StreamSlice] = None, | ||
| ) -> None: | ||
| path = [path.eval(self.config) for path in self._field_path] | ||
| if "*" in path: | ||
| matched = dpath.values(record, path) | ||
| extracted = matched[0] if matched else None | ||
| else: | ||
| extracted = dpath.get(record, path, default=[]) | ||
|
|
||
| if isinstance(extracted, dict): | ||
| conflicts = set(extracted.keys()) & set(record.keys()) | ||
| if not conflicts: | ||
| if self.delete_origin_value: | ||
| dpath.delete(record, path) | ||
| record.update(extracted) |
110 changes: 110 additions & 0 deletions
110
unit_tests/sources/declarative/transformations/test_dpath_flatten_fields.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,110 @@ | ||
| import pytest | ||
|
|
||
| from airbyte_cdk.sources.declarative.transformations.dpath_flatten_fields import DpathFlattenFields | ||
|
|
||
| _ANY_VALUE = -1 | ||
| _DELETE_ORIGIN_VALUE = True | ||
| _DO_NOT_DELETE_ORIGIN_VALUE = False | ||
|
|
||
|
|
||
| @pytest.mark.parametrize( | ||
| [ | ||
| "input_record", | ||
| "config", | ||
| "field_path", | ||
| "delete_origin_value", | ||
| "expected_record", | ||
| ], | ||
| [ | ||
| pytest.param( | ||
| {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}}, | ||
| {}, | ||
| ["field2"], | ||
| _DO_NOT_DELETE_ORIGIN_VALUE, | ||
| {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}, "field3": _ANY_VALUE}, | ||
| id="flatten by dpath, don't delete origin value", | ||
| ), | ||
| pytest.param( | ||
| {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}}, | ||
| {}, | ||
| ["field2"], | ||
| _DELETE_ORIGIN_VALUE, | ||
| {"field1": _ANY_VALUE, "field3": _ANY_VALUE}, | ||
| id="flatten by dpath, delete origin value", | ||
| ), | ||
| pytest.param( | ||
| { | ||
| "field1": _ANY_VALUE, | ||
| "field2": {"field3": {"field4": {"field5": _ANY_VALUE}}}, | ||
| }, | ||
| {}, | ||
| ["field2", "*", "field4"], | ||
| _DO_NOT_DELETE_ORIGIN_VALUE, | ||
| { | ||
| "field1": _ANY_VALUE, | ||
| "field2": {"field3": {"field4": {"field5": _ANY_VALUE}}}, | ||
| "field5": _ANY_VALUE, | ||
| }, | ||
| id="flatten by dpath with *, don't delete origin value", | ||
| ), | ||
| pytest.param( | ||
| { | ||
| "field1": _ANY_VALUE, | ||
| "field2": {"field3": {"field4": {"field5": _ANY_VALUE}}}, | ||
| }, | ||
| {}, | ||
| ["field2", "*", "field4"], | ||
| _DELETE_ORIGIN_VALUE, | ||
| {"field1": _ANY_VALUE, "field2": {"field3": {}}, "field5": _ANY_VALUE}, | ||
| id="flatten by dpath with *, delete origin value", | ||
| ), | ||
| pytest.param( | ||
| {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}}, | ||
| {"field_path": "field2"}, | ||
| ["{{ config['field_path'] }}"], | ||
| _DO_NOT_DELETE_ORIGIN_VALUE, | ||
| {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}, "field3": _ANY_VALUE}, | ||
| id="flatten by dpath from config, don't delete origin value", | ||
| ), | ||
| pytest.param( | ||
| {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}}, | ||
| {}, | ||
| ["non-existing-field"], | ||
| _DO_NOT_DELETE_ORIGIN_VALUE, | ||
| {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}}, | ||
| id="flatten by non-existing dpath, don't delete origin value", | ||
| ), | ||
| pytest.param( | ||
| {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}}, | ||
| {}, | ||
| ["*", "non-existing-field"], | ||
| _DO_NOT_DELETE_ORIGIN_VALUE, | ||
| {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}}, | ||
| id="flatten by non-existing dpath with *, don't delete origin value", | ||
| ), | ||
| pytest.param( | ||
| {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}, "field3": _ANY_VALUE}, | ||
| {}, | ||
| ["field2"], | ||
| _DO_NOT_DELETE_ORIGIN_VALUE, | ||
| {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}, "field3": _ANY_VALUE}, | ||
| id="flatten by dpath, not to update when record has field conflicts, don't delete origin value", | ||
| ), | ||
| pytest.param( | ||
| {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}, "field3": _ANY_VALUE}, | ||
| {}, | ||
| ["field2"], | ||
| _DO_NOT_DELETE_ORIGIN_VALUE, | ||
| {"field1": _ANY_VALUE, "field2": {"field3": _ANY_VALUE}, "field3": _ANY_VALUE}, | ||
| id="flatten by dpath, not to update when record has field conflicts, delete origin value", | ||
| ), | ||
| ], | ||
| ) | ||
| def test_dpath_flatten_lists( | ||
| input_record, config, field_path, delete_origin_value, expected_record | ||
| ): | ||
| flattener = DpathFlattenFields( | ||
| field_path=field_path, parameters={}, config=config, delete_origin_value=delete_origin_value | ||
| ) | ||
| flattener.transform(input_record) | ||
| assert input_record == expected_record |