-
Notifications
You must be signed in to change notification settings - Fork 0
/
hubspot.py
117 lines (80 loc) · 3 KB
/
hubspot.py
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
import json
import urllib
import requests
import config
BASE_URL = 'https://api.hubapi.com/'
class Contact(object):
data = {}
def __init__(self, data):
self.data = data
def get_email(self):
return next((item for item in self.data['identity-profiles'][0]['identities'] if item["type"] == "EMAIL"))['value']
def get_name(self):
properties_ = self.data['properties']
return properties_['firstname']['value'] + ' ' + properties_['lastname']['value']
def fetch_contacts():
contacts = []
offset = 0
has_more = True
while has_more:
url = get_url_to_fetch_contacts(offset)
response = requests.get(url)
response_data = response.content.decode('utf-8')
json_data = json.loads(response_data)
contacts_in_result = json_data['contacts']
contacts.extend([Contact(contact_data) for contact_data in contacts_in_result])
has_more = json_data['has-more']
offset = json_data['vid-offset']
return contacts
def get_url_to_fetch_contacts(offset, count=100):
params = {
'hapikey': config.HUBSPOT_HAPIKEY,
'count': count,
'vidOffset': offset,
}
return BASE_URL + 'contacts/v1/lists/all/contacts/all?%s' % urllib.parse.urlencode(params)
def find_contact_by_email(email):
url = get_url_to_find_contact_by_email(email)
response = requests.get(url)
if response.status_code == 404:
return None
response_data = response.content.decode('utf-8')
return Contact(json.loads(response_data))
def get_url_to_find_contact_by_email(email):
params = {
'hapikey': config.HUBSPOT_HAPIKEY
}
return BASE_URL + 'contacts/v1/contact/email/%s/profile?%s' % (email, urllib.parse.urlencode(params))
def create_or_update_contact(email, properties=None):
if properties is None:
properties = {}
data = {'properties': []}
for key in properties:
value = properties[key]
if value is not None:
data['properties'].append(get_property_dict(key, value))
url = get_url_to_create_or_update_contact(email)
response = requests.post(url, json=data)
def get_url_to_create_or_update_contact(email):
params = {'hapikey': config.HUBSPOT_HAPIKEY}
return BASE_URL + 'contacts/v1/contact/createOrUpdate/email/%s/?%s' % (email, urllib.parse.urlencode(params))
def get_property_dict(key, value):
return {
'property': key,
'value': value,
}
def find_contact(query):
contacts = find_contacts(query)
return next(contacts, None)
def find_contacts(query):
url = get_url_to_find_contacts(query)
response = requests.get(url)
response_data = response.content.decode('utf-8')
search_result = json.loads(response_data)
return [Contact(result) for result in search_result['contacts']]
def get_url_to_find_contacts(query):
params = {
'hapikey': config.HUBSPOT_HAPIKEY,
'q': query,
}
return BASE_URL + 'contacts/v1/search/query?%s' % urllib.parse.urlencode(params)