Build a HubSpot Deal Data API with Python and | Replit

Build a HubSpot Deal Data API with Python and Flask

A guide by

Horacio Lopez

Introduction

HubSpot is one of the leading CRMs used by companies of all sizes. HubSpot's API is great for pulling data for deals, contacts, and companies in your CRM. These associates can be visible within HubSpot but sometimes, you need to interact with this data programmatically with the associations merged in a clean, readable format.

This guide and the provided templates will show you how to fetch deals from HubSpot along with their associated contacts, companies, and owners, while respecting HubSpot's API rate limits. Once you get this associated data, you’ll be able to deploy this custom API as your own API for your team to use.

Using our templates, you can have your internal HubSpot Deals API up and running in 5 minutes or less.

Getting started

This guide comes with two sections:

  1. Creating the simple data aggregation script. Full code for this project here.
  2. Converting this script into an internal API. Full code for this project here.

For either template, click "Use template" and name your project however you like.

Create a Private app in HubSpot

Log in to your HubSpot account and navigate to Settings > Integrations > Private Apps

Click "Create a private app” and set a name for your app.

Set the necessary scopes (you'll need access to deals, contacts, companies, and owners).

Once your scopes are set, click Create app. You'll receive a private app access token.

To keep this secret safe, add it to the Secrets tab in your Repl. In the bottom-left corner of the Replit workspace, there is a section called "Tools."

Select "Secrets" within the Tools pane, and add your Private App access token to the Secret labeled HUBSPOT_ACCESS_TOKEN.

Setting up the HubSpot client and rate limiting

Let's start by setting up our HubSpot client and implementing a rate limiting system that keeps our API calls within HubSpot's rate limit of 10 calls per second:

import json
import os
import simplejson

from hubspot import HubSpot
from hubspot.crm.deals import ApiException
from urllib3.util.retry import Retry

# Initialize the HubSpot client with retry logic
hubspot_access_token = os.environ['HUBSPOT_ACCESS_TOKEN']
retry = Retry(total=5, status_forcelist=(429,))
hubspot_client = HubSpot(access_token=hubspot_access_token, retry=retry)

In Replit, os.environ['HUBSPOT_ACCESS_TOKEN'] will automatically fetch the token we stored in the Secrets tab.

Fetching deals

Now let's create a function to fetch deals:

def fetch_deals_with_associations():
    try:
        # Fetch 25 deals
        deals_page = hubspot_client.crm.deals.basic_api.get_page(limit=25)
        deals = deals_page.results

# Process each deal
        deals_with_associations = [process_deal(deal) for deal in deals]

# Log the response to the console
        print("Deals with associations have been fetched successfully")
        print(simplejson.dumps(deals_with_associations, indent=2, ignore_nan=True, default=str))
        return deals_with_associations

except ApiException as e:
        print(f"Exception when fetching deals: {e}\n")
        return f"Error: {e}\n"

This function fetches 25 deals, processes each one, and then prints the results using simplejson for better handling of complex data types. If you would like to pull more deals from your HubSpot CRM, simply increase the limit parameter.

Processing deals and fetching associated data

Let's create a function to process each deal and fetch its associated data:

def process_deal(deal):
    deal_dict = deal.to_dict()

try:
        # Get associated contacts
        contacts_page = hubspot_client.crm.associations.v4.basic_api.get_page(
            'deals', deal.id, 'contacts')
        contact_ids = [result.to_object_id for result in contacts_page.results]

# Get associated companies
        companies_page = hubspot_client.crm.associations.v4.basic_api.get_page(
            'deals', deal.id, 'companies')
        company_ids = [result.to_object_id for result in companies_page.results]

# Fetch owner details
        owner_email = None
        if deal.properties.get('hubspot_owner_id'):
            try:
                owner = hubspot_client.crm.owners.owners_api.get_by_id(
                    owner_id=deal.properties['hubspot_owner_id'],
                    id_property='id',
                    archived=False)
                owner_email = owner.properties.get('email')
            except ApiException as e:
                print(f"Exception when fetching owner details: {e}\n")

# Fetch full contact and company details
        contacts_batch = hubspot_client.crm.contacts.batch_api.read(
            batch_read_input_simple_public_object_id={
                "properties": ['email', 'firstname', 'lastname'],
                "inputs": [{"id": id} for id in contact_ids]
            }) if contact_ids else None

companies_batch = hubspot_client.crm.companies.batch_api.read(
            batch_read_input_simple_public_object_id={
                "properties": ['name', 'domain', 'hs_additional_domains', 'website'],
                "inputs": [{"id": id} for id in company_ids]
            }) if company_ids else None

# Format contact details
        formatted_contacts = []
        if contacts_batch and contacts_batch.results:
            for contact in contacts_batch.results:
                contact_dict = contact.to_dict()
                formatted_contacts.append({
                    "contact_id": int(contact_dict['id']),
                    "contact_email_addresses": [contact_dict['properties'].get('email')] if contact_dict['properties'].get('email') else [],
                    "contact_name": f"{contact_dict['properties'].get('firstname', '')} {contact_dict['properties'].get('lastname', '')}".strip()
                })

# Format company details
        formatted_companies = []
        if companies_batch and companies_batch.results:
            for company in companies_batch.results:
                company_dict = company.to_dict()
                primary_domain = company_dict['properties'].get('domain')
                additional_domains = company_dict['properties'].get('hs_additional_domains', '').split(';') if company_dict['properties'].get('hs_additional_domains') else []
                website = company_dict['properties'].get('website')
                all_domains = list(set(filter(None, [primary_domain, website] + additional_domains)))

formatted_companies.append({
                    "company_id": int(company_dict['id']),
                    "company_name": company_dict['properties'].get('name', ''),
                    "company_domains": all_domains
                })

# Combine deal with its associations and owner email
        return {
            **deal_dict,
            "owner_email": owner_email,
            "associatedContacts": formatted_contacts,
            "associatedCompanies": formatted_companies
        }

except ApiException as e:
        print(f"Exception when processing deal {deal.id}: {e}\n")
        return deal_dict

Running the script

To run our script, we add this at the end of the file:

# Run the function
fetch_deals_with_associations()

Once that’s in, hit the Run button in your Repl, if all is set up correctly, you will see your data appear in the Console.

Converting your script into an API

To make your HubSpot deal data aggregator more accessible to your team, let's convert it into a Flask API. This will allow other members of your organization to request deal data programmatically.

Setting up Flask

Let’s modify our script to convert it into a Flask application.:

app = Flask(__name__)

# Wrapper function to require API key
def require_api_key(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        api_key = request.headers.get('Authorization')
        if api_key and api_key == os.environ.get('INTERNAL_API_KEY'):
            return f(*args, **kwargs)
        else:
            return jsonify({"error": "Unauthorized"}), 401
    return decorated_function

@app.route('/get-deals-with-associations', methods=['GET'])
@require_api_key
def get_deals():
    deals = fetch_deals_with_associations()
    return jsonify(deals)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

Hit the Run button again. If everything is set up correctly, you should get the Webview tab to appear.

Securing your API

Next, we want to be sure that only members of our team will be able to send requests to our new API. Here’s how we can set up an internal key.

First, head to your Secrets tab again and add a new key called INTERNAL_API_KEY. For your value, add any password or secret string you like. Ideally,

you should generate a 15-20 character string with a mix of numbers, letters, and symbols. Paste this in as your value.

Testing your API

Once your Flask app is running, you can test it by using command:

curl -H "Authorization: YOUR_INTERNAL_API_KEY" YOUR_DEV_URL/get-deals-with-associations

Deploying your API

For APIs like this, we recommend using Autoscale. Click on Set up your deployment to continue.

On the next screen, you’ll be able to set your primary domain and edit the Secrets that will be in your production deployment. Usually, we keep these settings as they are.

Finally, click Deploy and watch your API go live!

Conclusion

You've now created a Flask API in Replit that fetches deals from HubSpot along with their associated contacts, companies, and owners. You can now integrate this API and data into other internal tools.