Step by Step Guide to Building an Apollo API Integration in Python

Aug 13, 2024 • 6 minute read

Introduction

Hey there, fellow developer! Ready to supercharge your prospecting game with Apollo's powerful API? In this guide, we'll walk through building a sleek Python integration that'll have you pulling valuable data in no time. Let's dive in!

Prerequisites

Before we get our hands dirty, make sure you've got:

Python 3.x installed
requests library (pip install requests)
An Apollo API key (if you don't have one, hop over to their website and grab it)

Setting up the project

Let's kick things off by creating a new Python file. I like to call mine apollo_integration.py, but feel free to get creative!

import requests
import json
import os

Authentication

First things first, let's keep that API key safe:

API_KEY = os.environ.get('APOLLO_API_KEY')
headers = {
    'Authorization': f'Bearer {API_KEY}',
    'Content-Type': 'application/json'
}

Pro tip: Never hardcode your API key. Use environment variables to keep it secure!

Making API requests

Now, let's craft our first request:

BASE_URL = 'https://api.apollo.io/v1'

def make_request(endpoint, method='GET', params=None, data=None):
    url = f'{BASE_URL}/{endpoint}'
    response = requests.request(method, url, headers=headers, params=params, json=data)
    response.raise_for_status()
    return response.json()

Implementing key Apollo API endpoints

Let's put our make_request function to work:

def search_contacts(query):
    return make_request('people/search', params={'q': query})

def get_account_info(domain):
    return make_request('organizations/enrich', params={'domain': domain})

def create_contact(data):
    return make_request('people', method='POST', data=data)

Error handling and rate limiting

Always be prepared for the unexpected:

from requests.exceptions import RequestException
import time

def rate_limited_request(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except RequestException as e:
            if e.response.status_code == 429:
                retry_after = int(e.response.headers.get('Retry-After', 60))
                print(f"Rate limited. Retrying after {retry_after} seconds.")
                time.sleep(retry_after)
                return func(*args, **kwargs)
            raise
    return wrapper

Data processing and storage

Let's make sense of that JSON:

import csv

def save_contacts_to_csv(contacts, filename='contacts.csv'):
    with open(filename, 'w', newline='') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=['name', 'email', 'company'])
        writer.writeheader()
        for contact in contacts:
            writer.writerow({
                'name': contact['name'],
                'email': contact['email'],
                'company': contact['organization']['name']
            })

Building a simple CLI interface

Time to make it interactive:

def main():
    while True:
        print("\n1. Search contacts\n2. Get account info\n3. Exit")
        choice = input("Enter your choice: ")
        if choice == '1':
            query = input("Enter search query: ")
            results = search_contacts(query)
            save_contacts_to_csv(results['people'])
            print(f"Saved {len(results['people'])} contacts to contacts.csv")
        elif choice == '2':
            domain = input("Enter domain: ")
            info = get_account_info(domain)
            print(json.dumps(info, indent=2))
        elif choice == '3':
            break
        else:
            print("Invalid choice. Try again.")

if __name__ == '__main__':
    main()

Best practices and optimization

Remember, API calls can be expensive. Cache results when possible and batch your requests when you can. Your future self (and your API quota) will thank you!

Testing the integration

Don't forget to test! Here's a quick example using unittest:

import unittest
from unittest.mock import patch

class TestApolloIntegration(unittest.TestCase):
    @patch('requests.request')
    def test_search_contacts(self, mock_request):
        mock_request.return_value.json.return_value = {'people': []}
        result = search_contacts('test query')
        self.assertEqual(result, {'people': []})

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

Conclusion

And there you have it! You've just built a robust Apollo API integration in Python. From authentication to data processing, you're now equipped to harness the power of Apollo's data in your Python projects.

Remember, this is just the beginning. Feel free to expand on this foundation, add more endpoints, or even build a full-fledged application. The sky's the limit!

Resources

Happy coding, and may your prospecting be ever fruitful!