<jupyter_start><jupyter_text>contacts> API details.<jupyter_code># hide
from nbdev.showdoc import *
# hide
# export

import time
import os
import json
import requests
from dotenv import load_dotenv
from am4894hubspot.search import hubspot_search
load_dotenv()
hapikey = os.environ.get('HUBSPOT_API_KEY','demo')
#hapikey = 'demo'
# export


def contact_exists(value, property='email', operator='EQ', hapikey=None) -> bool:
    """
    Check if a contact exists.
    """
    r = hubspot_search(value, property=property, operator=operator, hapikey=hapikey)
    if r.get('total',0) > 0:
        return True
    else:
        return False


# export


def get_contact(value, property='email', operator='EQ', hapikey=None, properties=None):
    """
    Get contact.
    """
    if hapikey is None:
        hapikey = os.environ.get('HUBSPOT_API_KEY','demo')
    r = hubspot_search(value, property=property, operator=operator, hapikey=hapikey, properties=properties)
    if 'total' in r:
        if r['total'] > 0:
            contact = r['results'][0]
            return contact
        else:
            return None
    else:
        return None

    
# export


def get_contact_by_id(id, hapikey=None, properties=None):
    """
    Get contact by id.
    """
    if hapikey is None:
        hapikey = os.environ.get('HUBSPOT_API_KEY','demo')
    endpoint = 'https://api.hubapi.com/crm/v3/objects/contacts/'
    url = f'{endpoint}/{id}'
    querystring = {"hapikey": hapikey, "properties": properties}
    headers = {"accept": "application/json"}
    response = requests.request("GET", url, headers=headers, params=querystring)
    
    return response


# export


def create_contact(contact=None, hapikey=None):
    """
    Create a contact.
    """
    if contact is None:
        contact = dict(email='<EMAIL>')
    if hapikey is None:
        hapikey = os.environ.get('HUBSPOT_API_KEY','demo')
    endpoint = 'https://api.hubapi.com/contacts/v1/contact/'
    url = f'{endpoint}?hapikey={hapikey}'
    headers = {
        "Content-Type": "application/json"
    }
    data = json.dumps({"properties": [{"property": k,"value": contact[k]} for k in contact]})
    r = requests.post(url=url, data=data, headers=headers)
    return r


# export 


def delete_contact(vid=None, hapikey=None):
    """
    Delete a contact.
    """
    if hapikey is None:
        hapikey = os.environ.get('HUBSPOT_API_KEY','demo')
    endpoint = f'https://api.hubapi.com/contacts/v1/contact/vid/{vid}'
    url = f'{endpoint}?hapikey={hapikey}'
    headers = {
        "Content-Type": "application/json"
    }
    r = requests.delete(url=url, headers=headers)
    return r


# export


def create_contacts(contacts=None, hapikey=None, check_exists=True):
    """
    Create contacts from a list of contacts.
    """
    if hapikey is None:
        hapikey = os.environ.get('HUBSPOT_API_KEY','demo')
    url = "https://api.hubapi.com/contacts/v1/contact/batch"
    querystring = {"hapikey": hapikey}
    headers = {"Content-Type": "application/json"}
    payload = []
    # loop over contacts and build payload to send
    for contact in contacts:
        email = contact["email"]
        payload.append({"email": email, "properties": [{"property": k,"value": contact[k]} for k in contact]})
    payload = json.dumps(payload)
    r = requests.request("POST", url, data=payload, headers=headers, params=querystring)
    if r.status_code == 202:
        return True
    else:
        raise ValueError(f"batch contact creation failed status_code={r.status_code} reason={r.reason}")
        return False


# export


def update_contact(email, properties, hapikey=None):
    """
    Update a contact.
    """
    if hapikey is None:
        hapikey = os.environ.get('HUBSPOT_API_KEY','demo')
    if contact_exists(email):
        contact = get_contact(email)
        vid = contact['id']
        endpoint = 'https://api.hubapi.com/contacts/v1/contact/'
        url = f'{endpoint}/vid/{vid}/profile?hapikey={hapikey}'
        headers = {
            "Content-Type": "application/json"
        }
        data = json.dumps({"properties": [{"property": k,"value": properties[k]} for k in properties]})
        r = requests.post(url=url, data=data, headers=headers)
        return r
    else:
        return None


# export


def create_or_update_contact(email, properties, hapikey=None):
    """
    Create or update a contact.
    """
    if hapikey is None:
        hapikey = os.environ.get('HUBSPOT_API_KEY','demo')
    endpoint = 'https://api.hubapi.com/contacts/v1/contact/'
    url = f'{endpoint}/createOrUpdate/email/{email}?hapikey={hapikey}'
    headers = {
        "Content-Type": "application/json"
    }
    data = json.dumps({"properties": [{"property": k,"value": properties[k]} for k in properties]})
    r = requests.post(url=url, data=data, headers=headers)
    return r


# hide
# tests

test_email = '<EMAIL>'

# first delete test contact if exists
contact = get_contact(test_email)
if contact:
    delete_contact(vid=contact['id'])

time.sleep(5)

# create an example minimal contact
contact_data = {
    "email": test_email
}

# create contact
response_create = create_contact(contact_data)
assert 'vid' in response_create.json()

time.sleep(5)

# check it exists
assert contact_exists(test_email) == True

time.sleep(2)

# get contact
contact = get_contact(test_email)
assert contact['properties']['email'] == test_email

time.sleep(2)

# update contact
response_update = update_contact(test_email, {"firstname": "fake", "lastname": "user"})
assert response_update.status_code == 204

time.sleep(5)

contact = get_contact(test_email)
assert contact['properties']['firstname'] == 'fake'
assert contact['properties']['lastname'] == 'user'

time.sleep(2)

# test create_or_update existing
response = create_or_update_contact(test_email, {"firstname": "fake_exists"}, hapikey)
assert response.json()['isNew'] == False

time.sleep(2)

# delete contact
response_delete = delete_contact(vid=contact['id'])
assert response_delete.status_code == 200

time.sleep(2)

# test create_or_update new
response = create_or_update_contact(test_email, {"firstname": "fake_new"}, hapikey)
assert response.json()['isNew'] == True
time.sleep(2)
contact = get_contact(test_email)
response_delete = delete_contact(vid=contact['id'])
assert response_delete.status_code == 200

## create batch of contacts
#test_contacts = [
#    dict(email='<EMAIL>'),
#    dict(email='<EMAIL>'),
#]
#assert create_contacts(test_contacts) == True
#time.sleep(5)
## now clean up and delete them
#for test_contact in test_contacts:
#    time.sleep(5)
#    test_email = test_contact['email']
#    test_contact_vid = get_contact(test_email)['id']
#    delete_contact(vid=test_contact_vid)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.14148290661536184, 'fraction_numerical': 0.00547580287109664, 'mean_word_length': 3.7028531663187194, 'pattern_counts': {'":': 27, '<': 9, '<?xml version=': 0, '>': 13, 'https://': 6, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '285203', 'license': 'Apache-2.0', 'n_tokens_mistral': 2242, 'n_tokens_neox': 2210, 'n_words': 485, 'path': '01_contacts.ipynb', 'repo_name': 'andrewm4894/am4894hubspot', 'size': '6808'}