-
Notifications
You must be signed in to change notification settings - Fork 72
/
engagements.py
101 lines (85 loc) · 3.43 KB
/
engagements.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
"""
hubspot engagements api
"""
from typing import Dict, List
from hubspot3.base import BaseClient
from hubspot3.utils import get_log
ENGAGEMENTS_API_VERSION = "1"
class EngagementsClient(BaseClient):
"""
The hubspot3 Engagements client uses the _make_request method to call the API
for data. It returns a python object translated from the json returned
"""
def __init__(self, *args, **kwargs) -> None:
super(EngagementsClient, self).__init__(*args, **kwargs)
self.log = get_log("hubspot3.engagements")
def _get_path(self, subpath: str) -> str:
"""get full subpath"""
return f"engagements/v{self.options.get('version') or ENGAGEMENTS_API_VERSION}/{subpath}"
def get(self, engagement_id, **options):
"""Get a HubSpot engagement."""
return self._call(f"engagements/{engagement_id}", method="GET", **options)
def get_associated(self, object_type, object_id, **options) -> List[Dict]:
"""
get all engagements associated with the given object
:param object_type: type of object to get associations on [CONTACT, COMPANY, DEAL]
:param object_id: ID of the object to get associations on
"""
finished = False
output = [] # type: List[Dict]
query_limit = 100 # Max value according to docs
offset = 0
while not finished:
batch = self._call(
f"engagements/associated/{object_type}/{object_id}/paged",
method="GET",
params={"limit": query_limit, "offset": offset},
**options,
)
output.extend(batch["results"])
finished = not batch["hasMore"]
offset = batch["offset"]
return output
def create(self, data=None, **options):
data = data or {}
return self._call("engagements", data=data, method="POST", **options)
def update(self, key, data=None, **options):
data = data or {}
return self._call(f"engagements/{key}", data=data, method="PUT", **options)
def patch(self, key, data=None, **options):
data = data or {}
return self._call(f"engagements/{key}", data=data, method="PATCH", **options)
def get_all(self, **options) -> List[Dict]:
"""get all engagements"""
finished = False
output = [] # type: List[Dict]
query_limit = 250 # Max value according to docs
offset = 0
while not finished:
batch = self._call(
"engagements/paged",
method="GET",
params={"limit": query_limit, "offset": offset},
**options,
)
output.extend(batch["results"])
finished = not batch["hasMore"]
offset = batch["offset"]
return output
def get_recently_modified(self, since, **options) -> List[Dict]:
"""get recently modified engagements"""
finished = False
output = [] # type: List[Dict]
query_limit = 100 # Max value according to docs
offset = 0
while not finished:
batch = self._call(
"engagements/recent/modified",
method="GET",
params={"limit": query_limit, "offset": offset, "since": since},
**options,
)
output.extend(batch["results"])
finished = not batch["hasMore"]
offset = batch["offset"]
return output