From 951e2195dce042c292cd96e494ae1f22fd81500e Mon Sep 17 00:00:00 2001 From: Alon Diamant Date: Wed, 14 Jun 2023 20:07:55 +0300 Subject: [PATCH] Added: CRM Association Labels (v4) API support Fix: mypy Fix: Prospector --- hubspot3/__init__.py | 7 + hubspot3/crm_association_labels.py | 252 +++++++++++++++++++++++++++++ hubspot3/test/test_init.py | 2 + 3 files changed, 261 insertions(+) create mode 100644 hubspot3/crm_association_labels.py diff --git a/hubspot3/__init__.py b/hubspot3/__init__.py index 76207ad..0123194 100644 --- a/hubspot3/__init__.py +++ b/hubspot3/__init__.py @@ -217,6 +217,13 @@ def crm_associations(self): return CRMAssociationsClient(**self.auth, **self.options) + @property + def crm_association_labels(self): + """returns a hubspot3 crm_association_labels client""" + from hubspot3.crm_association_labels import CRMAssociationLabelsClient + + return CRMAssociationLabelsClient(**self.auth, **self.options) + @property def crm_pipelines(self): """returns a hubspot3 crm_pipelines client""" diff --git a/hubspot3/crm_association_labels.py b/hubspot3/crm_association_labels.py new file mode 100644 index 0000000..87b6e18 --- /dev/null +++ b/hubspot3/crm_association_labels.py @@ -0,0 +1,252 @@ +""" +hubspot crm_association_labels api +""" +from enum import Enum +from typing import List, Dict, Optional, Union + +from hubspot3.base import BaseClient +from hubspot3.utils import get_log + +ASSOCIATIONS_API_VERSION = "4" + + +class ObjectTypeDefinitions(Enum): + """see https://developers.hubspot.com/docs/api/crm/understanding-the-crm""" + + CONTACTS = "0-1" + COMPANIES = "0-2" + DEALS = "0-3" + TICKETS = "0-5" + CALLS = "0-48" + EMAILS = "0-49" + MEETINGS = "0-47" + NOTES = "0-4" + TASKS = "0-27" + PRODUCTS = "0-7" + LINE_ITEMS = "0-8" + COMMUNICATIONS = "0-18" + POSTAL_MAIL = "0-116" + MARKETING_EVENTS = "0-54" + FEEDBACK_SUBMISSIONS = "0-19" + + +class AssociationCategory(Enum): + """see https://developers.hubspot.com/docs/cms/hubl/functions#crm-associations""" + + HUBSPOT_DEFINED = "HUBSPOT_DEFINED" + USER_DEFINED = "USER_DEFINED" + INTEGRATOR_DEFINED = "INTEGRATOR_DEFINED" + + +class CRMAssociationLabelsClient(BaseClient): + """ + Association labels extension for Associations API v4 endpoint + :see: https://developers.hubspot.com/docs/api/crm/associations + """ + + def __init__(self, *args, **kwargs): + super(CRMAssociationLabelsClient, self).__init__(*args, **kwargs) + self.log = get_log("hubspot3.crm_association_labels") + + def _get_path(self, subpath: str) -> str: + return f'crm/v{self.options.get("version") or ASSOCIATIONS_API_VERSION}/{subpath}' + + def list( + self, + from_object_type: ObjectTypeDefinitions, + from_object_id: int, + to_object_type: ObjectTypeDefinitions, + ) -> List[Dict]: + """ + List all association labels of an object by object type. + """ + finished = False + output = [] + after: Optional[str] = "" + # 100 is max value according to docs, but "You can only request at most 500 associations + # at once" error is thrown + query_limit = 500 + + while not finished: + params: Dict[str, Union[int, str]] = {"limit": query_limit} + if after: + params["after"] = after + batch = self._call( + f"objects/{from_object_type.value}/{from_object_id}/" + f"associations/{to_object_type.value}", + method="GET", + params=params, + ) + output.extend([id_ for id_ in batch["results"]]) + if ( + "paging" in batch + and "next" in batch["paging"] + and "after" in batch["paging"]["next"] + ): + after = batch["paging"]["next"]["after"] + else: + after = None + finished = not bool(after) + + return output + + def create_default( + self, + from_object_type: ObjectTypeDefinitions, + from_object_id: int, + to_object_type: ObjectTypeDefinitions, + to_object_id: int, + **options, + ) -> Dict: + """ + Create the default (most generic) association label between two object types + """ + return self._call( + f"objects/{from_object_type.value}/{from_object_id}/" + f"associations/default/{to_object_type.value}/{to_object_id}", + method="PUT", + **options, + ) + + def create( + self, + from_object_type: ObjectTypeDefinitions, + from_object_id: int, + to_object_type: ObjectTypeDefinitions, + to_object_id: int, + association_category: AssociationCategory, + association_type_id: int, + **options, + ) -> Dict: + """ + Set association labels between two records. + """ + return self._call( + f"objects/{from_object_type.value}/{from_object_id}/" + f"associations/{to_object_type.value}/{to_object_id}", + method="PUT", + data=[ + { + "associationCategory": association_category.value, + "associationTypeId": association_type_id, + } + ], + **options, + ) + + def delete( + self, + from_object_type: ObjectTypeDefinitions, + from_object_id: int, + to_object_type: ObjectTypeDefinitions, + to_object_id: int, + **options, + ) -> None: + """ + Deletes all association labels between two records. + """ + return self._call( + f"objects/{from_object_type.value}/{from_object_id}/associations" + f"/{to_object_type.value}/{to_object_id}", + method="DELETE", + **options, + ) + + def batch_list(self): + raise NotImplementedError("Not implemented yet") + + def batch_create_default(self): + raise NotImplementedError("Not implemented yet") + + def batch_create(self): + raise NotImplementedError("Not implemented yet") + + def batch_delete(self): + raise NotImplementedError("Not implemented yet") + + def batch_delete_specific(self): + raise NotImplementedError("Not implemented yet") + + def read_schema( + self, + from_object_type: ObjectTypeDefinitions, + to_object_type: ObjectTypeDefinitions, + ) -> List[Dict]: + """ + Returns all association label definitions between two object types + """ + results = [] + + response = self._call( + f"associations/{from_object_type.value}/{to_object_type.value}/labels", + method="GET", + ) + results.extend([id_ for id_ in response["results"]]) + + return results + + def create_schema( + self, + from_object_type: ObjectTypeDefinitions, + to_object_type: ObjectTypeDefinitions, + label: str, + name: str, + **options, + ) -> Dict: + """ + Create a user defined association label definition + """ + + response = self._call( + f"associations/{from_object_type.value}/{to_object_type.value}/labels", + method="POST", + data={ + "label": label, + "name": name, + }, + **options, + ) + if "results" in response and len(response) == 1: + return response["results"][0] + + raise RuntimeError( + f'Invalid response, results are not 1 item: {response["results"]}', + ) + + def update_schema( + self, + from_object_type: ObjectTypeDefinitions, + to_object_type: ObjectTypeDefinitions, + association_type_id: int, + label: str, + **options, + ) -> None: + """ + Update a user defined association label definition + """ + return self._call( + f"associations/{from_object_type.value}/{to_object_type.value}/labels", + method="PUT", + data={ + "label": label, + "associationTypeId": association_type_id, + }, + **options, + ) + + def delete_schema( + self, + from_object_type: ObjectTypeDefinitions, + to_object_type: ObjectTypeDefinitions, + association_type_id: int, + **options, + ) -> None: + """ + Deletes an association label definition + """ + return self._call( + f"associations/{from_object_type.value}/{to_object_type.value}" + f"/labels/{association_type_id}", + method="DELETE", + **options, + ) diff --git a/hubspot3/test/test_init.py b/hubspot3/test/test_init.py index b918e83..70e0f25 100644 --- a/hubspot3/test/test_init.py +++ b/hubspot3/test/test_init.py @@ -11,6 +11,7 @@ from hubspot3.contact_lists import ContactListsClient from hubspot3.contacts import ContactsClient from hubspot3.crm_associations import CRMAssociationsClient +from hubspot3.crm_association_labels import CRMAssociationLabelsClient from hubspot3.crm_pipelines import PipelinesClient from hubspot3.deals import DealsClient from hubspot3.engagements import EngagementsClient @@ -43,6 +44,7 @@ def test_create_hubspot3_client(): assert isinstance(hubspot.contact_lists, ContactListsClient) assert isinstance(hubspot.contacts, ContactsClient) assert isinstance(hubspot.crm_associations, CRMAssociationsClient) + assert isinstance(hubspot.crm_association_labels, CRMAssociationLabelsClient) assert isinstance(hubspot.crm_pipelines, PipelinesClient) assert isinstance(hubspot.deals, DealsClient) assert isinstance(hubspot.engagements, EngagementsClient)