-
Notifications
You must be signed in to change notification settings - Fork 3.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat[source-klaviyo]: added stream events detailed (#39350)
- Loading branch information
Showing
9 changed files
with
253 additions
and
10 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
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
57 changes: 57 additions & 0 deletions
57
...rations/connectors/source-klaviyo/source_klaviyo/components/inclouded_fields_extractor.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,57 @@ | ||
| # | ||
| # Copyright (c) 2024 Airbyte, Inc., all rights reserved. | ||
| # | ||
|
|
||
| from dataclasses import dataclass | ||
| from typing import Any, Iterable, Mapping | ||
|
|
||
| import dpath | ||
| import requests | ||
| from airbyte_cdk.sources.declarative.extractors.dpath_extractor import DpathExtractor | ||
|
|
||
|
|
||
| @dataclass | ||
| class KlaviyoIncludedFieldExtractor(DpathExtractor): | ||
| def extract_records(self, response: requests.Response) -> Iterable[Mapping[str, Any]]: | ||
| # Evaluate and retrieve the extraction paths | ||
| evaluated_field_paths = [field_path.eval(self.config) for field_path in self._field_path] | ||
| target_records = self.extract_records_by_path(response, evaluated_field_paths) | ||
| included_records = self.extract_records_by_path(response, ["included"]) | ||
|
|
||
| # Update target records with included records | ||
| updated_records = self.update_target_records_with_included(target_records, included_records) | ||
| yield from updated_records | ||
|
|
||
| @staticmethod | ||
| def update_target_records_with_included( | ||
| target_records: Iterable[Mapping[str, Any]], included_records: Iterable[Mapping[str, Any]] | ||
| ) -> Iterable[Mapping[str, Any]]: | ||
| for included_record in included_records: | ||
| included_attributes = included_record.get("attributes", {}) | ||
| for target_record in target_records: | ||
| target_relationships = target_record.get("relationships", {}) | ||
| included_record_type = included_record["type"] | ||
| if included_record_type in target_relationships: | ||
| target_relationships[included_record_type]["data"].update(included_attributes) | ||
| yield target_record | ||
|
|
||
| def extract_records_by_path(self, response: requests.Response, field_paths: list = None) -> Iterable[Mapping[str, Any]]: | ||
| response_body = self.decoder.decode(response) | ||
|
|
||
| # Extract data from the response body based on the provided field paths | ||
| if not field_paths: | ||
| extracted_data = response_body | ||
| else: | ||
| field_path_str = "/".join(field_paths) # Convert list of field paths to a single string path for dpath | ||
| if "*" in field_path_str: | ||
| extracted_data = dpath.util.values(response_body, field_path_str) | ||
| else: | ||
| extracted_data = dpath.util.get(response_body, field_path_str, default=[]) | ||
|
|
||
| # Yield extracted data as individual records | ||
| if isinstance(extracted_data, list): | ||
| yield from extracted_data | ||
| elif extracted_data: | ||
| yield extracted_data | ||
| else: | ||
| yield from [] |
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
59 changes: 59 additions & 0 deletions
59
airbyte-integrations/connectors/source-klaviyo/unit_tests/test_included_extractor.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,59 @@ | ||
| # | ||
| # Copyright (c) 2023 Airbyte, Inc., all rights reserved. | ||
| # | ||
|
|
||
| from unittest.mock import Mock, patch | ||
|
|
||
| import pytest | ||
| from requests.models import Response | ||
| from source_klaviyo.components.inclouded_fields_extractor import KlaviyoIncludedFieldExtractor | ||
|
|
||
|
|
||
| @pytest.fixture | ||
| def mock_response(): | ||
| return Mock(spec=Response) | ||
|
|
||
|
|
||
| @pytest.fixture | ||
| def mock_decoder(): | ||
| return Mock() | ||
|
|
||
|
|
||
| @pytest.fixture | ||
| def mock_config(): | ||
| return Mock() | ||
|
|
||
|
|
||
| @pytest.fixture | ||
| def mock_field_path(): | ||
| return [Mock() for _ in range(2)] | ||
|
|
||
|
|
||
| @pytest.fixture | ||
| def extractor(mock_config, mock_field_path, mock_decoder): | ||
| return KlaviyoIncludedFieldExtractor(mock_field_path, mock_config, mock_decoder) | ||
|
|
||
|
|
||
| @patch('dpath.util.get') | ||
| @patch('dpath.util.values') | ||
| def test_extract_records_by_path(mock_values, mock_get, extractor, mock_response, mock_decoder): | ||
| mock_values.return_value = [{'key': 'value'}] | ||
| mock_get.return_value = {'key': 'value'} | ||
| mock_decoder.decode.return_value = {'data': 'value'} | ||
|
|
||
| field_paths = ['data'] | ||
| records = list(extractor.extract_records_by_path(mock_response, field_paths)) | ||
| assert records == [{'key': 'value'}] | ||
|
|
||
| mock_values.return_value = [] | ||
| mock_get.return_value = None | ||
| records = list(extractor.extract_records_by_path(mock_response, ['included'])) | ||
| assert records == [] | ||
|
|
||
|
|
||
| def test_update_target_records_with_included(extractor): | ||
| target_records = [{'relationships': {'type1': {'data': {}}}}] | ||
| included_records = [{'type': 'type1', 'attributes': {'key': 'value'}}] | ||
|
|
||
| updated_records = list(extractor.update_target_records_with_included(target_records, included_records)) | ||
| assert updated_records[0]['relationships']['type1']['data'] == {'key': 'value'} |
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