hs_association_labels.py

from __future__ import annotations
from typing import *
from types import TracebackType
if TYPE_CHECKING:
    from requests import Request, Response

import json
import requests

from enum import Enum
from requests.adapters import HTTPAdapter
from requests.auth import AuthBase
from requests_toolbelt import sessions
from urllib3.util.retry import Retry


class StandardObjects(str, Enum):
    COMPANY   = 'companies'
    CONTACT   = 'contacts'
    DEAL      = 'deals'
    TICKET    = 'tickets'


class CustomObjects(str, Enum):
    """Define additional custom type names."""
    CUSTOM_OBJECT = '2-9184764'


class HubAPI(str, Enum):
    OBJECTS           = '/crm/v3/objects'
    SCHEMAS           = '/crm/v3/schemas'
    OWNERS            = '/crm/v3/owners'
    OBJECTSV4         = '/crm/v4/objects'
    ASSOCIATIONSV4    = '/crm/v4/associations'
    

class HubSpotAuth(AuthBase):

    def __init__(self, access_token: str) -> None:
        self._access_token = access_token

    def __call__(self, request: Request) -> Request:
        request.headers['Content-Type'] = 'application/json'
        request.headers['Authorization'] = f'Bearer {self._access_token}'
        return request


class HubSpotContext:

    def __init__(self, token: str = "") -> None:
        """
        HubSpot Session context manager.

        On enter, returns the session context.
        On exit, tears down the created context.

        Pass in a Private App access token to be consumed by a class that
        extends `requests.auth.AuthBase` to validate the connection.

        Parameters
        ----------
        token, optional
            HubSpot Private App access token, by default ""
        """
        self._base = 'https://api.hubapi.com'
        self._base_ctx = sessions.BaseUrlSession(base_url=self._base)
        self._base_ctx.auth = HubSpotAuth(token)
        self._base_ctx.mount(
            prefix=self._base,
            adapter=HTTPAdapter(
                max_retries=Retry(backoff_factor=1)
            )
        )

    def __enter__(self) -> requests.Session:
        return self._base_ctx

    def __exit__(
        self,
        exc_type: type[BaseException],
        exc_val: BaseException,
        exc_tb: TracebackType
    ) -> bool:
        self._base_ctx.close()
        return False


def create_association_label(
    token: str,
    from_obj: StandardObjects | CustomObjects, 
    to_obj: StandardObjects | CustomObjects,
    label: str,
    name: str,
) -> Response | None:
    resource_path = f'/{from_obj}/{to_obj}/labels'
    url = HubAPI.ASSOCIATIONSV4 + resource_path

    data = json.dumps({
        'label'   : label,
        'name'    : name,
    })
    
    with HubSpotContext(token) as context:
        response = context.post(url, data=data)
        return response