🎉 Source Mailchimp: support oauth flow (#7159)

* add mailchimp oauth support

* add PR

* fix creds

* upd spec

* format

* upd creds

* upd auth for different creds

* rename creds

* rename creds

* change ref in campaigns.json

* upd timeout_seconds

* merge

* add oauth java part

* add java test

* bump version

* update spec

* add anotation

* upd spec

* upd spec

* upd

* upd tests

* format

* upd

* upd

* add state

* add invalid_config

* bump version

* format

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -383,7 +383,7 @@
 - name: Mailchimp
   sourceDefinitionId: b03a9f3e-22a5-11eb-adc1-0242ac120002
   dockerRepository: airbyte/source-mailchimp
-  dockerImageTag: 0.2.10
+  dockerImageTag: 0.2.11
   documentationUrl: https://docs.airbyte.io/integrations/sources/mailchimp
   icon: mailchimp.svg
   sourceType: api

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -3764,31 +3764,109 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-mailchimp:0.2.10"
+- dockerImage: "airbyte/source-mailchimp:0.2.11"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/mailchimp"
     connectionSpecification:
       $schema: "http://json-schema.org/draft-07/schema#"
       title: "Mailchimp Spec"
       type: "object"
-      required:
-      - "username"
-      - "apikey"
-      additionalProperties: false
+      required: []
+      additionalProperties: true
       properties:
-        username:
-          type: "string"
-          title: "Username"
-          description: "The Username or email you use to sign into Mailchimp."
-        apikey:
-          type: "string"
-          airbyte_secret: true
-          title: "API Key"
-          description: "Mailchimp API Key. See the <a href=\"https://docs.airbyte.io/integrations/sources/mailchimp\"\
-            >docs</a> for information on how to generate this key."
+        credentials:
+          type: "object"
+          title: "Authentication Method"
+          oneOf:
+          - title: "OAuth2.0"
+            type: "object"
+            required:
+            - "auth_type"
+            - "access_token"
+            properties:
+              auth_type:
+                type: "string"
+                const: "oauth2.0"
+                enum:
+                - "oauth2.0"
+                default: "oauth2.0"
+                order: 0
+              client_id:
+                title: "Client ID"
+                type: "string"
+                description: "The Client ID of your OAuth application."
+                airbyte_secret: true
+              client_secret:
+                title: "Client Secret"
+                type: "string"
+                description: "The Client Secret of your OAuth application."
+                airbyte_secret: true
+              access_token:
+                title: "Access Token"
+                type: "string"
+                description: "An access token generated using the above client ID\
+                  \ and secret."
+                airbyte_secret: true
+          - type: "object"
+            title: "API Key"
+            required:
+            - "auth_type"
+            - "apikey"
+            properties:
+              auth_type:
+                type: "string"
+                const: "apikey"
+                enum:
+                - "apikey"
+                default: "apikey"
+                order: 1
+              apikey:
+                type: "string"
+                title: "API Key"
+                description: "Mailchimp API Key. See the <a href=\"https://docs.airbyte.io/integrations/sources/mailchimp\"\
+                  >docs</a> for information on how to generate this key."
+                airbyte_secret: true
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
+    advanced_auth:
+      auth_flow_type: "oauth2.0"
+      predicate_key:
+      - "credentials"
+      - "auth_type"
+      predicate_value: "oauth2.0"
+      oauth_config_specification:
+        complete_oauth_output_specification:
+          type: "object"
+          additionalProperties: false
+          properties:
+            access_token:
+              type: "string"
+              path_in_connector_config:
+              - "credentials"
+              - "access_token"
+        complete_oauth_server_input_specification:
+          type: "object"
+          additionalProperties: false
+          properties:
+            client_id:
+              type: "string"
+            client_secret:
+              type: "string"
+        complete_oauth_server_output_specification:
+          type: "object"
+          additionalProperties: false
+          properties:
+            client_id:
+              type: "string"
+              path_in_connector_config:
+              - "credentials"
+              - "client_id"
+            client_secret:
+              type: "string"
+              path_in_connector_config:
+              - "credentials"
+              - "client_secret"
 - dockerImage: "airbyte/source-mailgun:0.1.0"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/mailgun"

airbyte-integrations/connectors/source-mailchimp/Dockerfile

@@ -12,5 +12,5 @@ RUN pip install .
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.2.10
+LABEL io.airbyte.version=0.2.11
 LABEL io.airbyte.name=airbyte/source-mailchimp

airbyte-integrations/connectors/source-mailchimp/acceptance-test-config.yml

@@ -2,16 +2,41 @@ connector_image: airbyte/source-mailchimp:dev
 tests:
   spec:
     - spec_path: "source_mailchimp/spec.json"
+      timeout_seconds: 60
   connection:
+    # for old spec config (without oneOf)
     - config_path: "secrets/config.json"
       status: "succeed"
+      timeout_seconds: 180
+    # for auth with API token
+    - config_path: "secrets/config_apikey.json"
+      status: "succeed"
+      timeout_seconds: 180
+    # for auth with oauth2 token
+    - config_path: "secrets/config_oauth.json"
+      status: "succeed"
+      timeout_seconds: 180
     - config_path: "integration_tests/invalid_config.json"
       status: "failed"
+      timeout_seconds: 180
+    - config_path: "integration_tests/invalid_config_apikey.json"
+      status: "failed"
+      timeout_seconds: 180
+    - config_path: "integration_tests/invalid_config_oauth.json"
+      status: "failed"
+      timeout_seconds: 180
   discovery:
+    # for old spec config (without oneOf)
     - config_path: "secrets/config.json"
+    # for auth with API token
+    - config_path: "secrets/config_apikey.json"
+    # for auth with oauth2 token
+    - config_path: "secrets/config_oauth.json"
   basic_read:
     - config_path: "secrets/config.json"
       configured_catalog_path: "integration_tests/configured_catalog.json"
+    - config_path: "secrets/config_oauth.json"
+      configured_catalog_path: "integration_tests/configured_catalog.json"
 #  THIS TEST IS COMMENTED OUT. Tests are supposed to accept
 #  `state = {cursor_field: value}`. When we have dependent endpoint path
 #  `path_begin/{some_id}/path_end` we need a complex state like below:
@@ -30,3 +55,5 @@ tests:
   full_refresh:
     - config_path: "secrets/config.json"
       configured_catalog_path: "integration_tests/configured_catalog.json"
+    - config_path: "secrets/config_oauth.json"
+      configured_catalog_path: "integration_tests/configured_catalog.json"

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

@@ -0,0 +1,6 @@
+{
+  "credentials": {
+    "auth_type": "apikey",
+    "apikey": "api-key-awesome"
+  }
+}

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

@@ -0,0 +1,8 @@
+{
+  "credentials": {
+    "auth_type": "oauth2.0",
+    "client_id": "client_id",
+    "client_secret": "client_secret",
+    "access_token": "access_token"
+  }
+}

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

@@ -13,7 +13,6 @@
     packages=find_packages(),
     install_requires=[
         "airbyte-cdk~=0.1.35",
-        "mailchimp3==3.0.14",
         "pytest~=6.1",
     ],
     package_data={"": ["*.json", "schemas/*.json", "schemas/shared/*.json"]},

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

@@ -6,36 +6,61 @@
 import base64
 from typing import Any, List, Mapping, Tuple
 
+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 mailchimp3 import MailChimp
+from requests.auth import AuthBase
 
 from .streams import Campaigns, EmailActivity, Lists
 
 
-class HttpBasicAuthenticator(TokenAuthenticator):
-    def __init__(self, auth: Tuple[str, str], auth_method: str = "Basic", **kwargs):
-        # API keys have the format <key>-<data_center>.
-        # See https://mailchimp.com/developer/marketing/docs/fundamentals/#api-structure
-        self.data_center = auth[1].split("-").pop()
-        auth_string = f"{auth[0]}:{auth[1]}".encode("utf8")
-        b64_encoded = base64.b64encode(auth_string).decode("utf8")
-        super().__init__(token=b64_encoded, auth_method=auth_method, **kwargs)
+class MailChimpAuthenticator:
+    @staticmethod
+    def get_server_prefix(access_token: str) -> str:
+        try:
+            response = requests.get(
+                "https://login.mailchimp.com/oauth2/metadata", headers={"Authorization": "OAuth {}".format(access_token)}
+            )
+            return response.json()["dc"]
+        except Exception as e:
+            raise Exception(f"Cannot retrieve server_prefix for you account. \n {repr(e)}")
+
+    def get_auth(self, config: Mapping[str, Any]) -> AuthBase:
+        authorization = config.get("credentials", {})
+        auth_type = authorization.get("auth_type")
+        if auth_type == "apikey" or not authorization:
+            # API keys have the format <key>-<data_center>.
+            # See https://mailchimp.com/developer/marketing/docs/fundamentals/#api-structure
+            apikey = authorization.get("apikey") or config.get("apikey")
+            if not apikey:
+                raise Exception("No apikey in creds")
+            auth_string = f"anystring:{apikey}".encode("utf8")
+            b64_encoded = base64.b64encode(auth_string).decode("utf8")
+            auth = TokenAuthenticator(token=b64_encoded, auth_method="Basic")
+            auth.data_center = apikey.split("-").pop()
+
+        elif auth_type == "oauth2.0":
+            access_token = authorization["access_token"]
+            auth = TokenAuthenticator(token=access_token, auth_method="Bearer")
+            auth.data_center = self.get_server_prefix(access_token)
+
+        else:
+            raise Exception(f"Invalid auth type: {auth_type}")
+
+        return auth
 
 
 class SourceMailchimp(AbstractSource):
     def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) -> Tuple[bool, Any]:
         try:
-            client = MailChimp(mc_api=config["apikey"], mc_user=config["username"])
-            client.ping.get()
+            authenticator = MailChimpAuthenticator().get_auth(config)
+            requests.get(f"https://{authenticator.data_center}.api.mailchimp.com/3.0/ping", headers=authenticator.get_auth_header())
             return True, None
         except Exception as e:
             return False, repr(e)
 
     def streams(self, config: Mapping[str, Any]) -> List[Stream]:
-        authenticator = HttpBasicAuthenticator(auth=("anystring", config["apikey"]))
-        streams_ = [Lists(authenticator=authenticator), Campaigns(authenticator=authenticator), EmailActivity(authenticator=authenticator)]
-
-        return streams_
+        authenticator = MailChimpAuthenticator().get_auth(config)
+        return [Lists(authenticator=authenticator), Campaigns(authenticator=authenticator), EmailActivity(authenticator=authenticator)]