Hey there, fellow developer! Ready to supercharge your marketing automation with Klaviyo? Let's dive into building a robust Python integration with Klaviyo's API. This guide will walk you through the essentials, assuming you're already familiar with Python and API basics. We'll keep things concise and focused on what matters most.
Before we jump in, make sure you've got:
First things first, let's get our project structure in place:
mkdir klaviyo-integration cd klaviyo-integration python -m venv venv source venv/bin/activate # On Windows, use `venv\Scripts\activate` pip install requests
Klaviyo uses API keys for authentication. Let's set that up:
import os import requests KLAVIYO_API_KEY = os.getenv('KLAVIYO_API_KEY') BASE_URL = 'https://a.klaviyo.com/api/v2/' headers = { 'Accept': 'application/json', 'Content-Type': 'application/json', 'Api-Key': KLAVIYO_API_KEY }
Pro tip: Always use environment variables for sensitive info like API keys!
Let's start with some basic GET and POST requests:
def get_request(endpoint): response = requests.get(f"{BASE_URL}{endpoint}", headers=headers) response.raise_for_status() return response.json() def post_request(endpoint, data): response = requests.post(f"{BASE_URL}{endpoint}", headers=headers, json=data) response.raise_for_status() return response.json()
Klaviyo's API offers a wealth of endpoints. Here are some you'll likely use often:
lists
person
metric
event
def add_subscriber(list_id, email): endpoint = f"list/{list_id}/members" data = {"profiles": [{"email": email}]} return post_request(endpoint, data)
def update_profile(email, properties): endpoint = "person/update" data = {"email": email, "properties": properties} return post_request(endpoint, data)
def track_event(event_name, customer_properties, properties): endpoint = "track" data = { "event": event_name, "customer_properties": customer_properties, "properties": properties } return post_request(endpoint, data)
Klaviyo has rate limits, so let's implement some retry logic:
import time from requests.exceptions import RequestException def make_request_with_retry(func, *args, max_retries=3, **kwargs): for attempt in range(max_retries): try: return func(*args, **kwargs) except RequestException as e: if attempt == max_retries - 1: raise time.sleep(2 ** attempt) # Exponential backoff
For pagination, use the count
and page
parameters in your GET requests.
Unit testing is crucial. Here's a simple example:
import unittest from unittest.mock import patch class TestKlaviyoIntegration(unittest.TestCase): @patch('requests.post') def test_add_subscriber(self, mock_post): mock_post.return_value.json.return_value = {"id": "abc123"} result = add_subscriber("list_id", "[email protected]") self.assertEqual(result["id"], "abc123")
And there you have it! You've now got the building blocks for a solid Klaviyo API integration in Python. Remember, this is just the beginning – there's so much more you can do with Klaviyo's API. Keep exploring, keep coding, and most importantly, keep automating!
Happy coding, and may your email campaigns be ever successful!