✨ Source Mailchimp: Add optional start_date to config

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

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

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

@@ -4,13 +4,16 @@
 
 
 import base64
+import re
 from typing import Any, List, Mapping, Tuple
 
+import pendulum
 import requests
 from airbyte_cdk import AirbyteLogger
 from airbyte_cdk.sources import AbstractSource
 from airbyte_cdk.sources.streams import Stream
 from airbyte_cdk.sources.streams.http.auth import TokenAuthenticator
+from pendulum.parsing.exceptions import ParserError
 from requests.auth import AuthBase
 
 from .streams import (
@@ -78,7 +81,30 @@ def get_auth(self, config: Mapping[str, Any]) -> AuthBase:
 
 
 class SourceMailchimp(AbstractSource):
+    def _validate_start_date(self, config: Mapping[str, Any]):
+        start_date = config.get("start_date")
+
+        if start_date:
+            pattern = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z")
+            if not pattern.match(start_date):  # Compare against the pattern descriptor.
+                return "Please check the format of the start date against the pattern descriptor."
+
+            try:  # Handle invalid dates.
+                parsed_start_date = pendulum.parse(start_date)
+            except ParserError:
+                return "The provided start date is not a valid date. Please check the date you input and try again."
+
+            if parsed_start_date > pendulum.now("UTC"):  # Handle future start date.
+                return "The start date cannot be greater than the current date."
+
+        return None
+
     def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) -> Tuple[bool, Any]:
+        # First, check for a valid start date if it is provided
+        start_date_validation_error = self._validate_start_date(config)
+        if start_date_validation_error:
+            return False, start_date_validation_error
+
         try:
             authenticator = MailChimpAuthenticator().get_auth(config)
             response = requests.get(
@@ -102,21 +128,22 @@ 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")
+        start_date = config.get("start_date")
 
-        lists = Lists(authenticator=authenticator)
+        lists = Lists(authenticator=authenticator, start_date=start_date)
         interest_categories = InterestCategories(authenticator=authenticator, parent=lists)
 
         return [
-            Automations(authenticator=authenticator),
-            Campaigns(authenticator=authenticator),
-            EmailActivity(authenticator=authenticator, campaign_id=campaign_id),
+            Automations(authenticator=authenticator, start_date=start_date),
+            Campaigns(authenticator=authenticator, start_date=start_date),
+            EmailActivity(authenticator=authenticator, start_date=start_date, campaign_id=campaign_id),
             interest_categories,
             Interests(authenticator=authenticator, parent=interest_categories),
             lists,
-            ListMembers(authenticator=authenticator),
-            Reports(authenticator=authenticator),
-            SegmentMembers(authenticator=authenticator),
-            Segments(authenticator=authenticator),
+            ListMembers(authenticator=authenticator, start_date=start_date),
+            Reports(authenticator=authenticator, start_date=start_date),
+            SegmentMembers(authenticator=authenticator, start_date=start_date),
+            Segments(authenticator=authenticator, start_date=start_date),
             Tags(authenticator=authenticator, parent=lists),
-            Unsubscribes(authenticator=authenticator, campaign_id=campaign_id),
+            Unsubscribes(authenticator=authenticator, start_date=start_date, campaign_id=campaign_id),
         ]

airbyte-integrations/connectors/source-mailchimp/source_mailchimp/spec.json

@@ -61,6 +61,15 @@
           }
         ]
       },
+      "start_date": {
+        "title": "Incremental Sync Start Date",
+        "description": "The date from which you want to start syncing data for Incremental streams. Only records that have been created or modified since this date will be synced. If left blank, all data will by synced.",
+        "type": "string",
+        "format": "date-time",
+        "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}.[0-9]{3}Z$",
+        "pattern_descriptor": "YYYY-MM-DDTHH:MM:SS.000Z",
+        "examples": ["2020-01-01T00:00:00.000Z"]
+      },      
       "campaign_id": {
         "type": "string",
         "title": "ID of a campaign to sync email activities",

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

@@ -7,6 +7,7 @@
 from abc import ABC, abstractmethod
 from typing import Any, Iterable, List, Mapping, MutableMapping, Optional
 
+import pendulum
 import requests
 from airbyte_cdk.models import SyncMode
 from airbyte_cdk.sources.streams.core import StreamData
@@ -80,6 +81,10 @@ def read_records(
 class IncrementalMailChimpStream(MailChimpStream, ABC):
     state_checkpoint_interval = math.inf
 
+    def __init__(self, **kwargs):
+        self.start_date = kwargs.pop("start_date", None)
+        super().__init__(**kwargs)
+
     @property
     @abstractmethod
     def cursor_field(self) -> str:
@@ -124,6 +129,24 @@ def request_params(self, stream_state=None, stream_slice=None, **kwargs):
         params.update(default_params)
         return params
 
+    def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapping]:
+        """
+        Mailchimp endpoints do not always support filtering by date,
+        so we should filter out records manually against the start_date as a fallback.
+        """
+        response = super().parse_response(response, **kwargs)
+        start_date = pendulum.parse(self.start_date) if self.start_date else None
+
+        if start_date:
+            for record in response:
+                parsed_date = pendulum.parse(record.get(self.cursor_field))
+                if parsed_date >= start_date:
+                    yield record
+        # If no start date is provided, return all records
+        else:
+            for record in response:
+                yield record
+
 
 class MailChimpListSubStream(IncrementalMailChimpStream):
     """

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

@@ -14,7 +14,7 @@ def data_center_fixture():
 
 @fixture(name="config")
 def config_fixture(data_center):
-    return {"apikey": f"API_KEY-{data_center}"}
+    return {"apikey": f"API_KEY-{data_center}", "start_date": "2022-01-01T00:00:00.000Z"}
 
 
 @fixture(name="access_token")

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

@@ -88,6 +88,29 @@ def test_wrong_config(wrong_config):
         MailChimpAuthenticator().get_auth(wrong_config)
 
 
+@pytest.mark.parametrize(
+    "config, expected_return",
+    [
+        ({}, None),
+        ({"start_date": "2021-01-01T00:00:00.000Z"}, None),
+        ({"start_date": "2021-99-99T79:89:99.123Z"}, "The provided start date is not a valid date. Please check the date you input and try again."),
+        ({"start_date": "2021-01-01T00:00:00.000"}, "Please check the format of the start date against the pattern descriptor."),
+        ({"start_date": "2025-01-25T00:00:00.000Z"}, "The start date cannot be greater than the current date."),
+    ],
+    ids=[
+        "No start date",
+        "Valid start date",
+        "Invalid start date",
+        "Invalid format",
+        "Future start date",
+    ]
+)
+def test_validate_start_date(config, expected_return):
+    source = SourceMailchimp()
+    result = source._validate_start_date(config)
+    assert result == expected_return
+
+
 def test_streams_count(config):
     streams = SourceMailchimp().streams(config)
     assert len(streams) == 12

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

@@ -629,4 +629,29 @@ def test_reports_remove_empty_datetime_fields(auth, record, expected_return):
     """
     stream = Reports(authenticator=auth)
     assert stream.remove_empty_datetime_fields(record) == expected_return, f"Expected: {expected_return}, Actual: {stream.remove_empty_datetime_fields(record)}"
-
+
+
+@pytest.mark.parametrize("start_date, expected_ids", [
+        (None, [1, 2, 3]),
+        ("2022-01-01T00:00:00.000Z", [1, 3]),
+        ("2100-01-01T00:00:00.000Z", [])
+    ],
+    ids=[
+        "No start_date: all records read", 
+        "start_date provided: only records >= start_date read",
+        "Start date > records: no records read"
+    ]
+)
+def test_incremental_start_date(auth, start_date, expected_ids, requests_mock):
+    mock_data = [
+        {"id": 1, "create_time": "2022-01-05T00:00:00.000Z"},
+        {"id": 2, "create_time": "2021-12-25T00:00:00.000Z"},
+        {"id": 3, "create_time": "2022-01-01T00:00:00.000Z"},
+    ]
+    stream = Campaigns(authenticator=auth, start_date=start_date)
+
+    response = MagicMock()
+    response.json.return_value = {"campaigns": mock_data}
+    records = list(stream.parse_response(response))
+
+    assert [record["id"] for record in records] == expected_ids

docs/integrations/sources/mailchimp.md

@@ -13,12 +13,13 @@ You can use [OAuth](https://mailchimp.com/developer/marketing/guides/access-user
 3. On the Set up the source page, select **Mailchimp** from the Source type dropdown.
 4. Enter a name for your source.
 
-6. You can use OAuth or an API key to authenticate your Mailchimp account. We recommend using OAuth for Airbyte Cloud and an API key for Airbyte Open Source.
+5. You can use OAuth or an API key to authenticate your Mailchimp account. We recommend using OAuth for Airbyte Cloud and an API key for Airbyte Open Source.
     - To authenticate using OAuth for Airbyte Cloud, ensure you have [registered your Mailchimp account](#prerequisite) and then click **Authenticate your Mailchimp account** to sign in with Mailchimp and authorize your account. 
     - To authenticate using an API key for Airbyte Open Source, select **API key** from the Authentication dropdown and enter the [API key](https://mailchimp.com/developer/marketing/guides/quick-start/#generate-your-api-key) for your Mailchimp account.    
     :::note
     Check the [performance considerations](#performance-considerations) before using an API key.
     :::
+6. (Optional) You may optionally provide an **Incremental Sync Start Date** using the provided datepicker, or by programmatically entering a UTC date-time in the format `YYYY-MM-DDThh:mm:ss.sssZ`. If set, only data generated on or after the configured date-time will be synced for Incremental streams. Leaving this field blank will sync all data returned from the API. Please note that this option has no effect on streams using Full Refresh sync mode.
 7. Click **Set up source**.
 
 ## Supported sync modes
@@ -37,12 +38,17 @@ The Mailchimp source connector supports the following streams:
 [Email Activity](https://mailchimp.com/developer/marketing/api/email-activity-reports/list-email-activity/)
 [Interests](https://mailchimp.com/developer/marketing/api/interests/list-interests-in-category/)
 [Interest Categories](https://mailchimp.com/developer/marketing/api/interest-categories/list-interest-categories/)
+[Interests](https://mailchimp.com/developer/marketing/api/interests/list-interests-in-category/)
+[Interest Categories](https://mailchimp.com/developer/marketing/api/interest-categories/list-interest-categories/)
 [Lists](https://mailchimp.com/developer/api/marketing/lists/get-list-info)
 [List Members](https://mailchimp.com/developer/marketing/api/list-members/list-members-info/)
+[List Members](https://mailchimp.com/developer/marketing/api/list-members/list-members-info/)
 [Reports](https://mailchimp.com/developer/marketing/api/reports/list-campaign-reports/)
 [Segments](https://mailchimp.com/developer/marketing/api/list-segments/list-segments/)
 [Segment Members](https://mailchimp.com/developer/marketing/api/list-segment-members/list-members-in-segment/)
 [Tags](https://mailchimp.com/developer/marketing/api/lists-tags-search/search-for-tags-on-a-list-by-name/)
+[Segment Members](https://mailchimp.com/developer/marketing/api/list-segment-members/list-members-in-segment/)
+[Tags](https://mailchimp.com/developer/marketing/api/lists-tags-search/search-for-tags-on-a-list-by-name/)
 [Unsubscribes](https://mailchimp.com/developer/marketing/api/unsub-reports/list-unsubscribed-members/)
 
 ### A note on primary keys
@@ -78,6 +84,7 @@ Now that you have set up the Mailchimp source connector, check out the following
 
 | Version | Date       | Pull Request                                             | Subject                                                                    |
 |---------|------------|----------------------------------------------------------|----------------------------------------------------------------------------|
+| 0.11.0  | 2023-11-27 | [32852](https://github.com/airbytehq/airbyte/pull/32852) | Add optional start_date for incremental streams                            |
 | 0.10.0  | 2023-11-23 | [32782](https://github.com/airbytehq/airbyte/pull/32782) | Add SegmentMembers stream                                                  |
 | 0.9.0   | 2023-11-17 | [32218](https://github.com/airbytehq/airbyte/pull/32218) | Add Interests, InterestCategories, Tags streams                            |
 | 0.8.3   | 2023-11-15 | [32543](https://github.com/airbytehq/airbyte/pull/32543) | Handle empty datetime fields in Reports stream                             |