✨ Source Mailchimp: Add Interests, InterestCategories, Tags streams (#…

…32218)

airbyte-integrations/connectors/source-mailchimp/integration_tests/configured_catalog.json

@@ -28,6 +28,28 @@
       "primary_key": [["id"]],
       "destination_sync_mode": "append"
     },
+    {
+      "stream": {
+        "name": "interest_categories",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"],
+        "source_defined_primary_key": [["id"]]
+      },
+      "sync_mode": "full_refresh",
+      "primary_key": [["id"]],
+      "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "interests",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"],
+        "source_defined_primary_key": [["id"]]
+      },
+      "sync_mode": "full_refresh",
+      "primary_key": [["id"]],
+      "destination_sync_mode": "append"
+    },
     {
       "stream": {
         "name": "lists",
@@ -98,6 +120,17 @@
       "primary_key": [["id"]],
       "destination_sync_mode": "append"
     },
+    {
+      "stream": {
+        "name": "tags",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"],
+        "source_defined_primary_key": [["id"]]
+      },
+      "sync_mode": "full_refresh",
+      "primary_key": [["id"]],
+      "destination_sync_mode": "append"
+    },
     {
       "stream": {
         "name": "unsubscribes",

airbyte-integrations/connectors/source-mailchimp/integration_tests/configured_catalog_without_email_activities.json

@@ -28,6 +28,28 @@
       "primary_key": [["id"]],
       "destination_sync_mode": "append"
     },
+    {
+      "stream": {
+        "name": "interest_categories",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"],
+        "source_defined_primary_key": [["id"]]
+      },
+      "sync_mode": "full_refresh",
+      "primary_key": [["id"]],
+      "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "interests",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"],
+        "source_defined_primary_key": [["id"]]
+      },
+      "sync_mode": "full_refresh",
+      "primary_key": [["id"]],
+      "destination_sync_mode": "append"
+    },
     {
       "stream": {
         "name": "lists",
@@ -84,6 +106,17 @@
       "primary_key": [["id"]],
       "destination_sync_mode": "append"
     },
+    {
+      "stream": {
+        "name": "tags",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"],
+        "source_defined_primary_key": [["id"]]
+      },
+      "sync_mode": "full_refresh",
+      "primary_key": [["id"]],
+      "destination_sync_mode": "append"
+    },
     {
       "stream": {
         "name": "unsubscribes",

airbyte-integrations/connectors/source-mailchimp/metadata.yaml

@@ -10,7 +10,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: b03a9f3e-22a5-11eb-adc1-0242ac120002
-  dockerImageTag: 0.8.3
+  dockerImageTag: 0.9.0
   dockerRepository: airbyte/source-mailchimp
   documentationUrl: https://docs.airbyte.com/integrations/sources/mailchimp
   githubIssueLabel: source-mailchimp

airbyte-integrations/connectors/source-mailchimp/source_mailchimp/schemas/interest_categories.json

@@ -0,0 +1,21 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "type": "object",
+  "properties": {
+    "list_id": {
+      "type": ["null", "string"]
+    },
+    "id": {
+      "type": ["null", "string"]
+    },
+    "title": {
+      "type": ["null", "string"]
+    },
+    "display_order": {
+      "type": ["null", "integer"]
+    },
+    "type": {
+      "type": ["null", "string"]
+    }
+  }
+}

airbyte-integrations/connectors/source-mailchimp/source_mailchimp/schemas/interests.json

@@ -0,0 +1,24 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "type": "object",
+  "properties": {
+    "category_id": {
+      "type": ["null", "string"]
+    },
+    "list_id": {
+      "type": ["null", "string"]
+    },
+    "id": {
+      "type": ["null", "string"]
+    },
+    "name": {
+      "type": ["null", "string"]
+    },
+    "subscriber_count": {
+      "type": ["null", "string"]
+    },
+    "display_order": {
+      "type": ["null", "integer"]
+    }
+  }
+}

airbyte-integrations/connectors/source-mailchimp/source_mailchimp/schemas/tags.json

@@ -0,0 +1,15 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "type": "object",
+  "properties": {
+    "id": {
+      "type": ["null", "integer"]
+    },
+    "name": {
+      "type": ["null", "string"]
+    },
+    "list_id": {
+      "type": ["null", "string"]
+    }
+  }
+}

airbyte-integrations/connectors/source-mailchimp/source_mailchimp/source.py

@@ -13,7 +13,19 @@
 from airbyte_cdk.sources.streams.http.auth import TokenAuthenticator
 from requests.auth import AuthBase
 
-from .streams import Automations, Campaigns, EmailActivity, ListMembers, Lists, Reports, Segments, Unsubscribes
+from .streams import (
+    Automations,
+    Campaigns,
+    EmailActivity,
+    InterestCategories,
+    Interests,
+    ListMembers,
+    Lists,
+    Reports,
+    Segments,
+    Tags,
+    Unsubscribes,
+)
 
 
 class MailChimpAuthenticator:
@@ -89,13 +101,20 @@ def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) ->
     def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         authenticator = MailChimpAuthenticator().get_auth(config)
         campaign_id = config.get("campaign_id")
+
+        lists = Lists(authenticator=authenticator)
+        interest_categories = InterestCategories(authenticator=authenticator, parent=lists)
+
         return [
             Automations(authenticator=authenticator),
             Campaigns(authenticator=authenticator),
             EmailActivity(authenticator=authenticator, campaign_id=campaign_id),
-            Lists(authenticator=authenticator),
+            interest_categories,
+            Interests(authenticator=authenticator, parent=interest_categories),
+            lists,
             ListMembers(authenticator=authenticator),
             Reports(authenticator=authenticator),
             Segments(authenticator=authenticator),
+            Tags(authenticator=authenticator, parent=lists),
             Unsubscribes(authenticator=authenticator, campaign_id=campaign_id),
         ]

airbyte-integrations/connectors/source-mailchimp/source_mailchimp/streams.py

@@ -10,7 +10,7 @@
 import requests
 from airbyte_cdk.models import SyncMode
 from airbyte_cdk.sources.streams.core import StreamData
-from airbyte_cdk.sources.streams.http import HttpStream
+from airbyte_cdk.sources.streams.http import HttpStream, HttpSubStream
 
 logger = logging.getLogger("airbyte")
 
@@ -260,6 +260,53 @@ def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapp
                 yield {**item, **activity_item}
 
 
+class InterestCategories(MailChimpStream, HttpSubStream):
+    """
+    Get information about interest categories for a specific list.
+    Docs link: https://mailchimp.com/developer/marketing/api/interest-categories/list-interest-categories/
+    """
+
+    data_field = "categories"
+
+    def path(self, stream_slice: Mapping[str, Any] = None, **kwargs) -> str:
+        """
+        Get the list_id from the parent stream slice and use it to construct the path.
+        """
+        list_id = stream_slice.get("parent").get("id")
+        return f"lists/{list_id}/interest-categories"
+
+    def request_params(self, **kwargs):
+
+        # Exclude the _links field, as it is not user-relevant data
+        params = super().request_params(**kwargs)
+        params["exclude_fields"] = "categories._links"
+        return params
+
+
+class Interests(MailChimpStream, HttpSubStream):
+    """
+    Get a list of interests for a specific interest category.
+    Docs link: https://mailchimp.com/developer/marketing/api/interests/list-interests-in-category/
+    """
+
+    data_field = "interests"
+
+    def path(self, stream_slice: Mapping[str, Any] = None, **kwargs) -> str:
+        """
+        Get the list_id from the parent stream slice and use it to construct the path.
+        """
+        list_id = stream_slice.get("parent").get("list_id")
+        category_id = stream_slice.get("parent").get("id")
+        return f"lists/{list_id}/interest-categories/{category_id}/interests"
+
+    def request_params(self, **kwargs):
+
+        # Exclude the _links field, as it is not user-relevant data
+        params = super().request_params(**kwargs)
+        params["exclude_fields"] = "interests._links"
+        return params
+
+
 class ListMembers(MailChimpListSubStream):
     """
     Get information about members in a specific Mailchimp list.
@@ -308,6 +355,29 @@ class Segments(MailChimpListSubStream):
     data_field = "segments"
 
 
+class Tags(MailChimpStream, HttpSubStream):
+    """
+    Get information about tags for a specific list.
+    Docs link: https://mailchimp.com/developer/marketing/api/list-tags/list-tags-for-list/
+    """
+
+    data_field = "tags"
+
+    def path(self, stream_slice: Mapping[str, Any] = None, **kwargs) -> str:
+        list_id = stream_slice.get("parent").get("id")
+        return f"lists/{list_id}/tag-search"
+
+    def parse_response(self, response: requests.Response, stream_slice, **kwargs) -> Iterable[Mapping]:
+        """
+        Tags do not reference parent_ids, so we need to add the list_id to each record.
+        """
+        response = super().parse_response(response, **kwargs)
+
+        for record in response:
+            record["list_id"] = stream_slice.get("parent").get("id")
+            yield record
+
+
 class Unsubscribes(IncrementalMailChimpStream):
     """
     List of members who have unsubscribed from a specific campaign.

airbyte-integrations/connectors/source-mailchimp/unit_tests/test_source.py

@@ -90,4 +90,4 @@ def test_wrong_config(wrong_config):
 
 def test_streams_count(config):
     streams = SourceMailchimp().streams(config)
-    assert len(streams) == 8
+    assert len(streams) == 11