feat(cdp): add mailchimp template (#25781)

Co-authored-by: github-actions <41898282+github-actions[bot]@users.noreply.github.com>

posthog/cdp/templates/__init__.py

@@ -28,6 +28,7 @@
 from .activecampaign.template_activecampaign import template as activecampaign
 from .google_ads.template_google_ads import template as google_ads
 from .attio.template_attio import template as attio
+from .mailchimp.template_mailchimp import template as mailchimp
 from .microsoft_teams.template_microsoft_teams import template as microsoft_teams
 from .klaviyo.template_klaviyo import template_user as klaviyo_user, template_event as klaviyo_event
 from .google_cloud_storage.template_google_cloud_storage import (
@@ -59,6 +60,7 @@
     klaviyo_user,
     knock,
     loops,
+    mailchimp,
     mailgun,
     mailjet_create_contact,
     mailjet_update_contact_list,

posthog/cdp/templates/mailchimp/template_mailchimp.py

@@ -0,0 +1,135 @@
+from posthog.cdp.templates.hog_function_template import HogFunctionTemplate
+
+template: HogFunctionTemplate = HogFunctionTemplate(
+    status="beta",
+    id="template-mailchimp",
+    name="Mailchimp",
+    description="Updates a contact in Mailchimp and subscribes new ones.",
+    icon_url="/static/services/mailchimp.png",
+    category=["Email Marketing"],
+    hog="""
+if (empty(inputs.email)) {
+    print('No email set. Skipping...')
+    return
+}
+
+let properties := {}
+
+for (let key, value in inputs.properties) {
+    if (not empty(value)) {
+        properties[key] := value
+    }
+}
+
+if (inputs.include_all_properties) {
+    for (let key, value in event.properties) {
+        if (not empty(value) and not key like '$%') {
+            properties[key] := value
+        }
+    }
+}
+
+let userStatus := fetch(f'https://{inputs.dataCenterId}.api.mailchimp.com/3.0/lists/{inputs.audienceId}/members/{md5Hex(inputs.email)}', {
+    'method': 'GET',
+    'headers': {
+        'Authorization': f'Bearer {inputs.apiKey}',
+        'Content-Type': 'application/json'
+    }
+})
+
+if (userStatus.status == 404 or userStatus.status == 200) {
+    let res := fetch(f'https://{inputs.dataCenterId}.api.mailchimp.com/3.0/lists/{inputs.audienceId}/members/{md5Hex(inputs.email)}', {
+        'method': 'PUT',
+        'headers': {
+            'Authorization': f'Bearer {inputs.apiKey}',
+            'Content-Type': 'application/json'
+        },
+        'body': {
+            'email_address': inputs.email,
+            'status_if_new': inputs.doubleOptIn ? 'pending' : 'subscribed',
+            'merge_fields': properties
+        }
+    })
+    if (res.status >= 400) {
+        throw Error(f'Error from api.mailchimp.com (status {userStatus.status}): {userStatus.body}')
+    }
+} else if (userStatus.status >= 400) {
+    throw Error(f'Error from api.mailchimp.com (status {userStatus.status}): {userStatus.body}')
+}
+
+""".strip(),
+    inputs_schema=[
+        {
+            "key": "apiKey",
+            "type": "string",
+            "label": "Mailchimp API Key",
+            "description": "See the docs here: https://mailchimp.com/help/about-api-keys/",
+            "secret": True,
+            "required": True,
+        },
+        {
+            "key": "audienceId",
+            "type": "string",
+            "label": "Mailchimp audience ID",
+            "description": "See the docs here: https://mailchimp.com/help/find-audience-id/",
+            "secret": False,
+            "required": True,
+        },
+        {
+            "key": "dataCenterId",
+            "type": "string",
+            "label": "Mailchimp data center ID",
+            "description": "You can find your Datacenter ID in the Mailchimp url in your browser when you're logged in. It's the 'us1' in 'https://us1.admin.mailchimp.com/lists/'",
+            "secret": False,
+            "required": True,
+        },
+        {
+            "key": "email",
+            "type": "string",
+            "label": "Email of the user",
+            "description": "Where to find the email for the contact to be created. You can use the filters section to filter out unwanted emails or internal users.",
+            "default": "{person.properties.email}",
+            "secret": False,
+            "required": True,
+        },
+        {
+            "key": "doubleOptIn",
+            "type": "boolean",
+            "label": "Enable double opt-in",
+            "description": "If enabled, Mailchimp sends a confirmation email to that user, and that email is tagged with a pending subscriber status. The subscriber status automatically changes to subscribed once the user confirms the email.",
+            "default": False,
+            "secret": False,
+            "required": True,
+        },
+        {
+            "key": "include_all_properties",
+            "type": "boolean",
+            "label": "Include all event properties",
+            "description": "If set, all person properties will be included. Individual properties can be overridden below.",
+            "default": False,
+            "secret": False,
+            "required": True,
+        },
+        {
+            "key": "properties",
+            "type": "dictionary",
+            "label": "Merge field",
+            "description": "Map of Mailchimp merge fields and their values. You can use the filters section to filter out unwanted events. Check out this page for more details: https://mailchimp.com/developer/marketing/docs/merge-fields/#add-merge-data-to-contacts",
+            "default": {
+                "FNAME": "{person.properties.firstname}",
+                "LNAME": "{person.properties.lastname}",
+                "COMPANY": "{person.properties.company}",
+            },
+            "secret": False,
+            "required": False,
+        },
+    ],
+    filters={
+        "events": [
+            {"id": "$identify", "name": "$identify", "type": "events", "order": 0},
+            {"id": "$set", "name": "$set", "type": "events", "order": 1},
+        ],
+        "actions": [],
+        "filter_test_accounts": True,
+    },
+)

posthog/cdp/templates/mailchimp/test_template_mailchimp.py

@@ -0,0 +1,92 @@
+from inline_snapshot import snapshot
+from posthog.cdp.templates.helpers import BaseHogFunctionTemplateTest
+from posthog.cdp.templates.mailchimp.template_mailchimp import (
+    template as template_mailchimp,
+)
+
+
+def create_inputs(**kwargs):
+    inputs = {
+        "apiKey": "abcdef",
+        "audienceId": "a1b2c3",
+        "dataCenterId": "us1",
+        "email": "[email protected]",
+        "include_all_properties": False,
+        "doubleOptIn": False,
+        "properties": {"FNAME": "Max", "LNAME": "AI", "COMPANY": "PostHog"},
+    }
+    inputs.update(kwargs)
+
+    return inputs
+
+
+class TestTemplateMailchimp(BaseHogFunctionTemplateTest):
+    template = template_mailchimp
+
+    def test_function_works(self):
+        self.run_function(
+            inputs=create_inputs(),
+            globals={
+                "event": {"event": "$identify"},
+            },
+        )
+
+        assert self.get_mock_fetch_calls()[0] == snapshot(
+            (
+                "https://us1.api.mailchimp.com/3.0/lists/a1b2c3/members/12d91149f17f7ac265e833ea05ef6249",
+                {
+                    "method": "GET",
+                    "headers": {
+                        "Authorization": "Bearer abcdef",
+                        "Content-Type": "application/json",
+                    },
+                },
+            )
+        )
+
+    def test_body_includes_all_properties_if_set(self):
+        self.run_function(
+            inputs=create_inputs(include_all_properties=False),
+            globals={
+                "event": {"properties": {"PHONE": "+1415000000"}},
+            },
+        )
+
+        assert self.get_mock_fetch_calls()[1][1]["body"]["merge_fields"] == snapshot(
+            {"FNAME": "Max", "LNAME": "AI", "COMPANY": "PostHog"}
+        )
+
+        self.run_function(
+            inputs=create_inputs(include_all_properties=True),
+            globals={
+                "event": {"properties": {"PHONE": "+1415000000"}},
+            },
+        )
+
+        assert self.get_mock_fetch_calls()[1][1]["body"]["merge_fields"] == snapshot(
+            {
+                "FNAME": "Max",
+                "LNAME": "AI",
+                "COMPANY": "PostHog",
+                "PHONE": "+1415000000",
+            }
+        )
+
+    def test_double_opt_in(self):
+        self.run_function(
+            inputs=create_inputs(doubleOptIn=False),
+        )
+
+        assert self.get_mock_fetch_calls()[1][1]["body"]["status_if_new"] == snapshot("subscribed")
+
+        self.run_function(
+            inputs=create_inputs(doubleOptIn=True),
+        )
+
+        assert self.get_mock_fetch_calls()[1][1]["body"]["status_if_new"] == snapshot("pending")
+
+    def test_function_requires_identifier(self):
+        self.run_function(inputs=create_inputs(email=""))
+
+        assert not self.get_mock_fetch_calls()
+        assert self.get_mock_print_calls() == snapshot([("No email set. Skipping...",)])