I have a Python script to add a note to all of my contacts that have “Active” status and the role of “Primary Owner”. The script retrieves the contacts fine; however, it fails when it tries to create a note for each contact. I’ve used Notes - v3 | HubSpotAPI and Engagements | Notes | HubSpot for references.
Code:
from hubspot import HubSpot
from hubspot.crm.contacts import SimplePublicObjectInput, ApiException
from hubspot.crm.objects import ApiException
from hubspot.crm.objects.notes import SimplePublicObjectInputForCreate, ApiException
from datetime import datetime as d
import datetime as dt
import logging as log
log.basicConfig(level=log.INFO, format='%(asctime)s | %(levelname)s | %(message)s', datefmt='%m-%d-%Y %H:%M:%S')
log.info('Initializing...')
API_KEY = 'REDACTED'
BASE_URL = 'https://api.hubapi.com'
STATUS_PROP = 'hs_content_membership_status'
STATUS_VALUE = 'active'
ROLE_PROP = 'role'
ROLE_VALUE = 'Primary Owner'
ERROR_STATUS = False
client = HubSpot(access_token=API_KEY)
log.info('Connected to HubSpot')
# Create a note (engagement) to add to the contacts
def create_note_for_contact(contact_id, note_content):
properties = {
"hs_note_body": f"{str(note_content)}",
"hs_timestamp": f"{str(d.now(dt.timezone.utc).timestamp() * 1000)}",
"hubspot_owner_id": "REDACTED",
"hs_attachment_ids": "REDACTED"
}
simple_public_object_input_for_create = SimplePublicObjectInputForCreate(
associations=[{"types":[{"associationCategory":"HUBSPOT_DEFINED","associationTypeId":202}],
"to":{contact_id}}],
# object_write_trace_id='string',
properties=properties
)
try:
response = client.crm.objects.notes.basic_api.create(simple_public_object_input_for_create=simple_public_object_input_for_create)
log.info(f"Note added to contact {contact_id}")
except ApiException as e:
log.error(f"Error adding note to contact {contact_id}: {e}")
global ERROR_STATUS
ERROR_STATUS = True
# Function to get contacts with 'hs_content_membership_status' = 'active'
def get_active_contacts():
filter_criteria = {
"filterGroups": [{
"filters": [{
"propertyName": f"{STATUS_PROP}",
"operator": "EQ",
"value": f"{STATUS_VALUE}"
},
{
"propertyName": f"{ROLE_PROP}",
"operator": "EQ",
"value": f"{ROLE_VALUE}"
}]
}],
"limit": 100,
"properties": ["firstname", "lastname", "hs_content_membership_status"]
}
try:
# Fetch contacts with filter
response = client.crm.contacts.search_api.do_search(filter_criteria)
contacts = response.results
log.info(f"Found {len(contacts)} contacts with active membership status.")
return contacts
except ApiException as e:
log.error(f"Error fetching contacts: {e}")
global ERROR_STATUS
ERROR_STATUS = True
return []
# Main function to add a note to all active contacts
def add_note_to_all_active_contacts(note_content):
contacts = get_active_contacts()
try:
log.info(f"Adding note to {len(contacts)} contacts...")
for contact in contacts:
contact_id = contact.id
create_note_for_contact(contact_id, note_content)
except ApiException as e:
log.error(f"Error adding note to contacts: {e}")
global ERROR_STATUS
ERROR_STATUS = True
def confirm_input():
valid_responses = ['y', 'n', 'yes', 'no']
while True:
response = input('Are you sure you want to continue? (y/n): ')
if response in valid_responses:
if response.lower() in ['y', 'yes']:
return True
else:
return False
else:
log.error('Invalid response. Please enter "y" or "n".')
if __name__ == "__main__":
log.info('Starting...')
# Example note content
note = input('Enter the note to be added to all active primary contacts: \n')
if confirm_input():
log.info(f'User confirmed adding the following note to all active primary contacts:\n {note}')
# Run the function to add a note to all active contacts
add_note_to_all_active_contacts(note)
if ERROR_STATUS:
log.error('Error(s) occurred. Please check the logs.')
else:
log.info('Done!')
else:
log.warning('User cancelled the operation.')
The error I have been getting is:
Mark this answer as solution if it helped you