Build a HubSpot Deal Data API with Python and Flask
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:
- Creating the simple data aggregation script. Full code for this project here.
- 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 HubSpots rate limit of 10 calls per second:
python
1
2
3
4
5
6
7
8
9
10
11
12
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:
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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 in this line:
deals_page = hubspot_client.crm.deals.basic_api.get_page(limit=25)
Processing deals and fetching associated data #
Let's create a function to process each deal and fetch its associated data:
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
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_dictTip: If you want a deeper explanation of what a certain section of code does in your project, simply highlight the lines in question, right click, and select “Explain with AI”.
Running the script #
To run our script, we add this at the end of the file:
python
1
2
# 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. From here you can ask Replit AI to turn the results into a CSV, txt file, or continue adding more functionality once you receive your aggregated Deals data.
If you’d like to turn this into an internal API that the rest of your team can use, read on!
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. At the bottom of main.py replace the fetch_deals_with_associations() function with the following:
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
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)Add these imports at the top:
python
1
2
from flask import Flask, jsonify, request
from functools import wrapsYour final main.py file should be:
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
import os
from functools import wraps
import simplejson
from flask import Flask, jsonify, request
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)
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"
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
# Run the flask app
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 like this:
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.
You can store this value in a secure password manager like 1Password to share with the rest of your team.
Testing your API #
Once your Flask app is running, it will automatically have its own development environment URL. To find it, click on the Webview URL preview it should show a pop up like this:
You can also find your development URL by typing ⌘ + K (or Ctrl + K) and typing "networking". Copy this URL to your clipboard.
Then, open the Shell and enter the following command. Replace YOUR_INTERNAL_API_KEY with the value in your Secrets and YOUR_DEV_URL with the URL copied from the Webview or Networking tab.
python
1
curl -H "Authorization: YOUR_INTERNAL_API_KEY" YOUR_DEV_URL/get-deals-with-associationsDeploying your API #
Open a new tab in the Workspace and search for “Deployments” or open the control console by typing ⌘ + K (or Ctrl + K) and type "deploy". You should find a screen like this.
For APIs like this, we recommend using Autoscale. Click on Set up your deployment to continue. On the next screen, click Approve and configure build settings most internal APIs work fine with the default machine settings but if you need more power later, you can always come back and change these settings later. You can monitor your usage and billing at any time at: replit.com/usage.
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!
If your deployment succeeds, you now have a production URL and /get-deals-with-associations endpoint for the rest of your team to pull clean, readable data directly from your HubSpot CRM.
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 (e.g. Salesforce or Google Sheets).
This API respects HubSpot's rate limit of 10 requests per second limits, provides a comprehensive view of your deal data, and offers a secure way for your team to access this data programmatically.