🎉 Source Amazon Ads : use optional config report_record_types

airbyte-integrations/connectors/source-amazon-ads/integration_tests/spec.json

@@ -72,6 +72,26 @@
         "examples": [3, 10],
         "order": 8,
         "type": "integer"
+      },
+      "report_record_types": {
+        "title": "Report Record Types",
+        "description": "Optional configuration which accepts an array of string of record types. Leave blank for default behaviour to pull all report types. Use this config option only if you want to pull specific report type(s). See <a href=\"https://advertising.amazon.com/API/docs/en-us/reporting/v2/report-types\">docs</a> for more details",
+        "items": {
+          "type": "string",
+          "enum": [
+            "adGroups",
+            "asins",
+            "asins_keywords",
+            "asins_targets",
+            "campaigns",
+            "keywords",
+            "productAds",
+            "targets"
+          ]
+        },
+        "type": "array",
+        "uniqueItems": true,
+        "order": 9
       }
     },
     "required": ["client_id", "client_secret", "refresh_token"],

airbyte-integrations/connectors/source-amazon-ads/source_amazon_ads/spec.yaml

@@ -86,6 +86,28 @@ connectionSpecification:
       type: "integer"
       default: 3
       order: 8
+    report_record_types:
+      title: Report Record Types
+      description:
+        Optional configuration which accepts an array of string of record types.
+        Leave blank for default behaviour to pull all report types.
+        Use this config option only if you want to pull specific report type(s).
+        See <a href="https://advertising.amazon.com/API/docs/en-us/reporting/v2/report-types">docs</a>
+        for more details
+      items:
+        type: string
+        enum:
+          - adGroups
+          - asins
+          - asins_keywords
+          - asins_targets
+          - campaigns
+          - keywords
+          - productAds
+          - targets
+      type: array
+      uniqueItems: true
+      order: 9
   required:
     - client_id
     - client_secret

airbyte-integrations/connectors/source-amazon-ads/source_amazon_ads/streams/report_streams/report_streams.py

@@ -122,6 +122,8 @@ def __init__(self, config: Mapping[str, Any], profiles: List[Profile], authentic
         # Maximum retries Airbyte will attempt for fetching report data. Default is 5.
         self.report_generation_maximum_retries: int = get_typed_env("REPORT_GENERATION_MAX_RETRIES", 5)
 
+        self._report_record_types = config.get("report_record_types", [])
+
     @property
     def model(self) -> CatalogModel:
         return self._model
@@ -364,6 +366,9 @@ def _init_reports(self, profile: Profile, report_date: str) -> List[ReportInfo]:
         """
         report_infos = []
         for record_type, metrics in self.metrics_map.items():
+            if len(self._report_record_types) > 0 and record_type not in self._report_record_types:
+                continue
+
             report_init_body = self._get_init_report_body(report_date, record_type, profile)
             if not report_init_body:
                 continue

airbyte-integrations/connectors/source-amazon-ads/unit_tests/test_report_streams.py

@@ -486,7 +486,6 @@ def test_read_incremental_with_records(config):
     stream = SponsoredDisplayReportStream(config, profiles, authenticator=mock.MagicMock())
 
     with freeze_time("2021-01-02 12:00:00") as frozen_datetime:
-
         state = {}
         records = list(read_incremental(stream, state))
         assert state == {"1": {"reportDate": "20210102"}}
@@ -531,7 +530,6 @@ def test_read_incremental_without_records_start_date(config):
     stream = SponsoredDisplayReportStream(config, profiles, authenticator=mock.MagicMock())
 
     with freeze_time("2021-01-02 12:00:00") as frozen_datetime:
-
         state = {}
         reportDates = ["20201231", "20210101", "20210102", "20210103", "20210104"]
         for reportDate in reportDates:
@@ -554,7 +552,6 @@ def test_read_incremental_with_records_start_date(config):
     stream = SponsoredDisplayReportStream(config, profiles, authenticator=mock.MagicMock())
 
     with freeze_time("2021-01-02 12:00:00") as frozen_datetime:
-
         state = {}
         records = list(read_incremental(stream, state))
 
@@ -643,3 +640,148 @@ def test_streams_state_filter(mocker, config, state_filter, stream_class):
         assert params["stateFilter"] == ",".join(state_filter)
     else:
         assert state_filter is None
+
+
+@responses.activate
+@pytest.mark.parametrize(
+    "custom_record_types, flag_match_error",
+    [
+        (
+                ["campaigns"],
+                True
+        ),
+        (
+                ["campaigns", "adGroups"],
+                True
+        ),
+        (
+                [],
+                False
+        ),
+        (
+                ["invalid_record_type"],
+                True
+        )
+    ]
+)
+def test_display_report_stream_with_custom_record_types(config_gen, custom_record_types, flag_match_error):
+    setup_responses(
+        init_response=REPORT_INIT_RESPONSE,
+        status_response=REPORT_STATUS_RESPONSE,
+        metric_response=METRIC_RESPONSE,
+    )
+
+    profiles = make_profiles()
+
+    stream = SponsoredDisplayReportStream(config_gen(report_record_types=custom_record_types), profiles, authenticator=mock.MagicMock())
+    stream_slice = {"profile": profiles[0], "reportDate": "20210725"}
+    records = list(stream.read_records(SyncMode.incremental, stream_slice=stream_slice))
+    for record in records:
+        if record['recordType'] not in custom_record_types:
+            if flag_match_error:
+                assert False
+
+
+@responses.activate
+@pytest.mark.parametrize(
+    "custom_record_types, expected_record_types, flag_match_error",
+    [
+        (
+                ["campaigns"],
+                ["campaigns"],
+                True
+        ),
+        (
+                ["asins_keywords"],
+                ["asins_keywords"],
+                True
+        ),
+        (
+                ["asins_targets"],
+                ["asins_targets"],
+                True
+        ),
+        (
+                ["campaigns", "adGroups"],
+                ["campaigns", "adGroups"],
+                True
+        ),
+        (
+                [],
+                [],
+                False
+        ),
+        (
+                ["invalid_record_type"],
+                [],
+                True
+        )
+    ]
+)
+def test_products_report_stream_with_custom_record_types(config_gen, custom_record_types, expected_record_types, flag_match_error):
+    setup_responses(
+        init_response_products=REPORT_INIT_RESPONSE,
+        status_response=REPORT_STATUS_RESPONSE,
+        metric_response=METRIC_RESPONSE,
+    )
+
+    profiles = make_profiles(profile_type="vendor")
+
+    stream = SponsoredProductsReportStream(config_gen(report_record_types=custom_record_types), profiles, authenticator=mock.MagicMock())
+    stream_slice = {"profile": profiles[0], "reportDate": "20210725", "retry_count": 3}
+    records = list(stream.read_records(SyncMode.incremental, stream_slice=stream_slice))
+    for record in records:
+        print(record)
+        if record['recordType'] not in expected_record_types:
+            if flag_match_error:
+                assert False
+
+
+@responses.activate
+@pytest.mark.parametrize(
+    "custom_record_types, expected_record_types, flag_match_error",
+    [
+        (
+                ["campaigns"],
+                ["campaigns"],
+                True
+        ),
+        (
+                ["asins"],
+                ["asins"],
+                True
+        ),
+        (
+                ["campaigns", "adGroups"],
+                ["campaigns", "adGroups"],
+                True
+        ),
+        (
+                [],
+                [],
+                False
+        ),
+        (
+                ["invalid_record_type"],
+                [],
+                True
+        )
+    ]
+)
+def test_brands_video_report_with_custom_record_types(config_gen, custom_record_types, expected_record_types, flag_match_error):
+    setup_responses(
+        init_response_brands=REPORT_INIT_RESPONSE,
+        status_response=REPORT_STATUS_RESPONSE,
+        metric_response=METRIC_RESPONSE,
+    )
+
+    profiles = make_profiles()
+
+    stream = SponsoredBrandsVideoReportStream(config_gen(report_record_types=custom_record_types), profiles, authenticator=mock.MagicMock())
+    stream_slice = {"profile": profiles[0], "reportDate": "20210725"}
+    records = list(stream.read_records(SyncMode.incremental, stream_slice=stream_slice))
+    for record in records:
+        print(record)
+        if record['recordType'] not in expected_record_types:
+            if flag_match_error:
+                assert False

docs/integrations/sources/amazon-ads.md

@@ -94,6 +94,7 @@ Information about expected report generation waiting time you may find [here](ht
 
 | Version | Date       | Pull Request                                             | Subject                                                                                                         |
 |:--------|:-----------|:---------------------------------------------------------|:----------------------------------------------------------------------------------------------------------------|
+| 1.0.1   | 2022-11-01 | [18677](https://github.com/airbytehq/airbyte/pull/18677) | Add optional config report_record_types                                                                         |
 | 1.0.0   | 2023-01-30 | [21677](https://github.com/airbytehq/airbyte/pull/21677) | Fix bug with non-unique primary keys in report streams. Add asins_keywords and asins_targets                    |
 | 0.1.29  | 2023-01-27 | [22038](https://github.com/airbytehq/airbyte/pull/22038) | Set `AvailabilityStrategy` for streams explicitly to `None`                                                     |
 | 0.1.28  | 2023-01-18 | [19491](https://github.com/airbytehq/airbyte/pull/19491) | Add option to customize look back window value                                                                  |