前提

Python/Django で API の開発をしている。
ここで、HubSpot という CRM を利用したい。やりたいことは以下のような感じ。

アプリケーション上でユーザーが会員登録した際、HubSpot 側にもユーザーデータを作成したい
ユーザーデータとは、会員情報･企業情報など

HubSpot のアカウント自体は作成済み。
おそらく、 HubSpot API を利用することになる。
が、そもそも CRM を利用した経験があまりない & ネット上に HubSpot の情報があまりないので、スクラップに作業過程を記載していく。

tenkei

API アクセストークンの取得

HubSpot API を利用するには、アクセストークンが必要。

アクセストークンを取得するにはまず、非公開アプリを作成すればOK。

非公開アプリとは、HubSpot アカウントに閉じた概念。いわゆる SaaS でプロジェクトなどと呼ばれているようなものだと思われる。
作成した非公開アプリは、同一の HubSpot アカウント内のメンバー全員が利用可能になる。

非公開アプリについて: https://developers.hubspot.jp/docs/api/private-apps

一方、公開アプリとは外部に公開されるもの。
おそらくこちらにリストされる → https://ecosystem.hubspot.com/ja/marketplace/apps
自社サービスのために HubSpot API を使いたい今回のような場合、公開アプリを作成する必要はなさそう。

ただし、公開アプリならタイムラインイベントAPI が利用可能となる。(詳細不明)
また、非公開アプリの方が Webhook の制限がある。(詳細不明)

他のトークンとしては、OAuth トークンと開発者アカウント API キーというものもある。
OAuth は個人の認証に紐づく。これは公開アプリのためのもの？(きちんと調査できていない)
開発者アカウント API キーもよく分からないが、今回は非公開アプリのアクセストークンを使えばやりたいことはできそう。

tenkei

Python SDK を利用する

PythonからHubSpot APIを利用するための公式SDKが用意されている。
→ https://github.com/HubSpot/hubspot-api-python

作業手順は README の通り。
SDKをインストールして、あとは任意の HubSpot API を利用する。

以下のコードは、作成済みの contact を取得する適当なコード。

import os

from hubspot import HubSpot


class HubSpotUtil:
    api_client = HubSpot(access_token=os.getenv('HUB_SPOT_ACCESS_TOKEN')) # 環境変数から渡す

    @classmethod
    def get_contact_by_id(cls, id: str):
        contact = cls.api_client.crm.contacts.basic_api.get_by_id(id).to_dict()
        print(contact) # degug
        return contact

無事にデータが取れていることを確認できた。

tenkei

API レート制限

API レート制限について。

非公開アプリから実行できる呼び出しの回数は、アカウントのご契約内容とAPI追加オプションのご購入状況に基づきます。
https://developers.hubspot.jp/docs/api/usage-details

非公開アプリの一覧から、API制限回数を確認できる。
自分たちの場合、1日に500,000件だと言う事がわかった。

tenkei

ローカル開発

HubSpot API を使う場合、ローカル開発環境はどうするか。

Sandbox 環境は提供されているが、バジェット的に利用しない方針に。
https://www.hubspot.jp/products/sandbox

その場合の方策として、もう1つ HubSpot アカウントを作成することにした。

それぞれの HubSpot アカウントで払い出したアクセストークンを、本番環境と開発環境で使い分ければ OK。

また、開発環境を｢開発者アカウント｣にすることで、開発がしやすくなるかもしれない。
開発者アカウントでは、開発者テストアカウントというものを複数作成できる。1つ1つの開発者テストアカウントが、別々のCRM空間を持つ。つまり、アカウントごとに Contect, Company, Deal などを分けられる、と認識した。

まだ試していないが、Local, Dev, Staging などの開発者テストアカウントを作るとやりやすいかもしれない。

tenkei

Contact, Company, Deal の取得と作成

Contact, Company, Deal の取得と作成は、README のような感じで実装できる。

from hubspot import HubSpot
from hubspot.crm.contacts import SimplePublicObjectInputForCreate as ContactInput
from hubspot.crm.companies import SimplePublicObjectInputForCreate as CompanyInput

api_client = HubSpot(access_token="Access Token")

contact_properties={
    "email": "test@example.com",
    "firstname": "Foo",
    "lastname": "Bar".

    # ...その他のプロパティたち
}
api_client.crm.contacts.basic_api.create(
    simple_public_object_input_for_create=ContactInput(properties=contact_properties)
)

company_properties = {
    "name": "Example Corporation",

    # ...その他のプロパティたち
}
api_client.crm.companies.basic_api.create(
    simple_public_object_input_for_create=CompanyInput(properties=company_properties)
)

# Deal も同様な感じなので省略

simple_public_object_input_for_create みたいな長過ぎる名前はもうちょっと何とかならなかったのか…と思わないでもない。

tenkei

Association の作成（= レコード同士の紐づけ）

レコード同士を紐づけるには、 Association API の Create Default エンドポイントを利用する。

Association API Create Default - developers.hubspot.com

例えば、 ID:100 の Contact を ID:1 の Company に紐づけるには、以下のように書ける。

from hubspot import HubSpot

api_client = HubSpot(access_token="Access Token")

api_client.crm.associations.v4.basic_api.create_default(
    from_object_type="contacts",
    from_object_id="100",
    to_object_type="companies",
    to_object_id="1",
)

相互の紐づけなので、 from と to を入れ替えても結果は同じっぽい。また、object_type の値を "deals"のように変えれば、紐づく対象の種別を変更できる。

作成に成功した際のレスポンスは以下のような感じ。

{
    "completed_at": "2024-07-11T09:23:31.943000Z",
    "num_errors": null,
    "requested_at": null,
    "started_at": "2024-07-11T09:23:31.870000Z",
    "links": null,
    "results": [
        {
            "association_spec": {
                "association_category": "HUBSPOT_DEFINED",
                "association_type_id": 280
            },
            "_from": {
                "id": "xxxxxxxxxxx"
            },
            "to": {
                "id": "xxxxxxxxxxx"
            }
        },
        {
            "association_spec": {
                "association_category": "HUBSPOT_DEFINED",
                "association_type_id": 279
            },
            "_from": {
                "id": "xxxxxxxxxxx"
            },
            "to": {
                "id": "xxxxxxxxxxx"
            }
        }
    ],
    "errors": null,
    "status": "COMPLETE"
    }
}

何やら association_type_id のようなものが払い出されている。

ちなみに、上記の Create Default エンドポイントの他に、 Create というエンドポイントもある。

Association API Create - developers.hubspot.com

こちらは、自分で association_type_id を指定する必要がありそう。なので、ほとんどのケースでは、Create Default エンドポイントを利用しておけば良さそうに思える。（Create エンドポイントはどんな場合に使うんだろう…?）

tenkei

ドメインによる Company と Contact の自動紐づけ

Company レコードに domain 値を保存すれば、そのドメインを含む email を持った Contact を自動紐づけしてくれる。

例えば、 Company の domain が example.com だとする。
新しく foo@example.com というメールアドレスを持った Contact を登録すると、勝手に Association が作成される。

ちなみに、 HubSpot の設定画面からその設定を ON にする必要はある。

参考: https://knowledge.hubspot.com/object-settings/automatically-create-and-associate-companies-with-contacts

tenkei

メール配信設定の更新

Subscriptions だか SubscriptionPreferences だかの API を使う。

前者が v4 と書いてあり、現時点での最新っぽい。が、Python SDK が対応してないっぽいので、後者を利用した。

が、後者のコードサンプルをそのまま実行してもエラーに…。
subscription_status_api を status_api に直したら動いた。

from hubspot import HubSpot
from hubspot.communication_preferences import PublicUpdateSubscriptionStatusRequest

api_client = HubSpot(access_token="Access Token")

public_update_subscription_status_request = PublicUpdateSubscriptionStatusRequest(
    email_address=email,
    subscription_id="ここにID",
    legal_basis="PERFORMANCE_OF_CONTRACT",
    legal_basis_explanation="ここに法的根拠"
)
api_client.communication_preferences.status_api.subscribe(
    public_update_subscription_status_request=public_update_subscription_status_request
)

subscription_id は適宜調べて埋める。
Subscriptions を取得する API があるので、その取得結果に含まれる ID を設定すると良いかもしれない。

コードを見て分かる通り、その他の API に比べて legal な感じ。 legal_basis とか legal_basis_explanation を適切に選択･入力する必要があるかも。

ちなみに、 legal_basis の enum はこんな感じ。

 [
    "LEGITIMATE_INTEREST_PQL",
    "LEGITIMATE_INTEREST_CLIENT",
    "PERFORMANCE_OF_CONTRACT",
    "CONSENT_WITH_NOTICE",
    "NON_GDPR",
    "PROCESS_AND_STORE",
    "LEGITIMATE_INTEREST_OTHER",
]

規約同意によって配信ONにするような場合、 PERFORMANCE_OF_CONTRACT かな。ユーザー自身がON/OFFにする場合は CONSENT_WITH_NOTICE が設定される。

tenkei

レコードの検索

Contact, Company, Deal などのレコードを検索したい場合。README の通りに実装すれば OK。

from hubspot import HubSpot
from hubspot.crm.companies import PublicObjectSearchRequest

api_client = HubSpot(access_token="Access Token")

public_object_search_request = PublicObjectSearchRequest(
    filter_groups=[
        {
            "filters": [
                {
                    "value": "1234567",
                    "propertyName": "zip",
                    "operator": "EQ"
                }
            ]
        }
    ], limit=10
)

api_response = api_client.crm.companies.search_api.do_search(public_object_search_request=public_object_search_request)

なんか独自のクエリ言語みたいなのを指定する。

Not Found の場合はエラーが返るとかではなく、空配列を含む正常レスポンスが返る。

tenkei

Association によるレコードの検索

association による検索は、 propertyName で指定すればよい。

以下は、特定の company に紐づく deal を取得している。

from hubspot import HubSpot
from hubspot.crm.deals import PublicObjectSearchRequest

api_client = HubSpot(access_token="Access Token")

public_object_search_request = PublicObjectSearchRequest(
    filter_groups=[
        {
            "filters": [
                {
                    "value": "1234567", # ← company_id
                    "propertyName": "associations.company",
                    "operator": "EQ"
                }
            ]
        }
    ],
)

api_response = api_client.crm.deals.search_api.do_search(public_object_search_request=public_object_search_request)

tenkei

エラーハンドリング

HubSpot API 連携時にエラーが発生した場合のハンドリング。

Error クラス内に status, reason, body などが含まれている。エラー通知などを行う場合、それらの情報を取れば良い。

Contact、Company, Deal などのエンドポイントでこのようなエラーが返ってくることを確認済み。

try:
    # Contact の作成や Association の作成など
except Exeption as e:
        status = getattr(e, 'status', "unknown")
        reason = getattr(e, 'reason', "unknown")
        body = getattr(e, 'body', "unknown")
        return f"エラーが発生しました！ Status: {status} {reason} Message: {body}"

tenkei

Deal(取引)のパイプラインとステージを指定する

Deal 作成時に pipeline と dealstage を指定すれば OK。

from hubspot import HubSpot
from hubspot.crm.deals import SimplePublicObjectInputForCreate

api_client = HubSpot(access_token="Access Token")

properties={
    "name": "取引A",
    "pipeline": "PIPELINE_ID", 
    "dealstage": "STAGE_ID".

    # ...その他のプロパティたち
}
api_client.crm.deals.basic_api.create(
    simple_public_object_input_for_create=SimplePublicObjectInputForCreate(properties=properties)
)

当たり前ですが、自作したパイプラインやステージを指定したい場合は、HubSpot側であらかじめそれらを作成しておく必要があります。

dealstageを指定しないと pipeline も設定されない、という問題があるので注意。