Step by Step Guide to Building a Hubspot Marketing Hub API Integration in Python

Introduction

Hey there, fellow developer! Ready to dive into the world of Hubspot Marketing Hub API integration? You're in for a treat. This guide will walk you through creating a robust Python integration that'll have you pulling data like a pro in no time.

Prerequisites

Before we jump in, make sure you've got:

• A Python environment (3.7+ recommended)
• A Hubspot account with API access
• Your favorite code editor

Got all that? Great! Let's get started.

Setting up the project

First things first, let's set up our project:

mkdir hubspot-integration
cd hubspot-integration
python -m venv venv
source venv/bin/activate  # On Windows, use `venv\Scripts\activate`
pip install requests python-dotenv

Authentication

Hubspot uses API keys for authentication. Let's keep it secure:

1. Create a .env file in your project root:

   HUBSPOT_API_KEY=your_api_key_here
   

2. Now, let's create an authentication function:

import os
from dotenv import load_dotenv

load_dotenv()

def get_headers():
    return {
        'Authorization': f"Bearer {os.getenv('HUBSPOT_API_KEY')}",
        'Content-Type': 'application/json'
    }

Making API requests

Time to make our first request! Here's a basic structure:

import requests

BASE_URL = "https://api.hubapi.com/crm/v3"

def get_contacts(limit=100):
    endpoint = f"{BASE_URL}/objects/contacts"
    params = {"limit": limit}
    response = requests.get(endpoint, headers=get_headers(), params=params)
    response.raise_for_status()
    return response.json()

Implementing key Marketing Hub API endpoints

Let's implement some crucial endpoints:

def get_lists():
    endpoint = f"{BASE_URL}/lists"
    response = requests.get(endpoint, headers=get_headers())
    response.raise_for_status()
    return response.json()

def create_email_campaign(name, subject, html_content):
    endpoint = f"{BASE_URL}/marketing/v3/transactional/single-email/send"
    payload = {
        "name": name,
        "subject": subject,
        "htmlContent": html_content
    }
    response = requests.post(endpoint, headers=get_headers(), json=payload)
    response.raise_for_status()
    return response.json()

# Add more endpoints as needed

Error handling and rate limiting

Let's add some retry logic and respect those rate limits:

from time import sleep
from requests.exceptions import RequestException

def api_call_with_retry(func, max_retries=3, delay=1):
    for attempt in range(max_retries):
        try:
            return func()
        except RequestException as e:
            if attempt == max_retries - 1:
                raise
            sleep(delay * (2 ** attempt))  # Exponential backoff

Data processing and storage

Now, let's process and store our data:

import csv

def save_contacts_to_csv(contacts, filename='contacts.csv'):
    with open(filename, 'w', newline='') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=['id', 'email', 'firstname', 'lastname'])
        writer.writeheader()
        for contact in contacts['results']:
            writer.writerow({
                'id': contact['id'],
                'email': contact['properties'].get('email', ''),
                'firstname': contact['properties'].get('firstname', ''),
                'lastname': contact['properties'].get('lastname', '')
            })

Building a simple use case

Let's put it all together with a simple example:

def sync_contacts_and_create_list():
    contacts = get_contacts(limit=1000)
    save_contacts_to_csv(contacts)
    
    list_name = "New Contacts List"
    list_id = create_list(list_name)
    
    for contact in contacts['results']:
        add_contact_to_list(contact['id'], list_id)
    
    print(f"Synced {len(contacts['results'])} contacts and added them to '{list_name}'")

# Don't forget to implement create_list and add_contact_to_list functions!

Testing and debugging

Always test your code! Here's a simple unit test:

import unittest

class TestHubspotIntegration(unittest.TestCase):
    def test_get_contacts(self):
        contacts = get_contacts(limit=1)
        self.assertIsNotNone(contacts)
        self.assertIn('results', contacts)
        self.assertEqual(len(contacts['results']), 1)

if __name__ == '__main__':
    unittest.main()

Best practices and optimization

Remember to:

• Use pagination for large datasets
• Batch your API calls when possible
• Keep your integration up-to-date with Hubspot's latest API changes

Conclusion

And there you have it! You've just built a Hubspot Marketing Hub API integration in Python. Pretty cool, right? Remember, this is just the beginning. There's so much more you can do with the Hubspot API. Keep exploring, keep coding, and most importantly, have fun!

For more information, check out the Hubspot API documentation. Happy coding!