【Python】MarketoのデータをAPIを使用して取得する方法

やったこと

Pythonを利用して、MarketoのスマートリストをAPIを使用してCSV形式でダウンロードする

用語の整理

・Python
プログラミング言語。記法も難しくなく、初心者にもやさしい。ライブラリも豊富で主にデータ処理や機械学習まわりなどに強い言語。

Welcome to Python.org

・Marketo
Adobe Marketo Engage。Adobe社が提供しているMAツール。

https://business.adobe.com/jp/products/marketo/adobe-marketo.html

・スマートリスト
Marketo内の機能の一つ。わかりやすくいうと動的グルーピングみたいな機能。

スマートリストについて | Adobe Marketo Engage

実装手順

①MarketoのRESTAPIを使用するための準備

以下の公式ドキュメントに従ってREST APIが利用できる環境を準備する。
やっていることは
・API用のユーザーの作成
・クライアントID/トークン発行
・エンドポイントの確認
▼公式ドキュメント

REST API - Marketo Developers

②取得したいデータを格納するスマートリストを作成

こちらもドキュメントに従い、スマートリストを作成する。
▼公式ドキュメント

スマートリストの作成 | Adobe Marketo Engage

その後、スマートリストの画面を開き、スマートリストタブにて取得したいデータの条件を設定（フィルタリング）する。
その際、URL欄に記載されているスマートリストIDをメモっておく
※スマートリストID
URL末尾の「SL●●●●」の●●●●の部分

③PythonでAPIを実行するためのコードを記載する

今回使用した、REST APIのメソッドは
・GET /identity/oauth/token
トークンを発行。有効期限があるので、実行の都度発行する。

・POST /bulk/v1/leads/export/create.json
一括抽出のためのジョブを作成。以下全てに言えるが、今回スマートリストを抽出したいとなった場合は一括抽出（Bulk）を使用する。

・POST /bulk/v1/leads/export/{exportId}/enqueue.json
一括抽出ジョブを実行。

・GET /bulk/v1/leads/export/{exportId}/status.json
ジョブのステータス取得。ジョブには時間がかかるものもあり、実行の完了を確認した上で次の処理に移すためにステータスを取得する。

・GET /bulk/v1/leads/export/{exportId}/file.json
ファイルを取得。つまり該当のジョブで取得したスマートリストのデータをCSV形式で取得。

各それぞれの詳細は下記をご参照ください。

Lead Database - Marketo Developers

ソースコード一部抜粋と簡単な解説

1.認証情報を変数として定義

# 認証情報(Marketo)
marketo_url = ★エンドポイント★
client_id = ★クライアントID★
client_secret = ★クライアントシークレット★

2.処理ごとに関数定義

2-1.認証トークン発行

# (Marketo)認証トークンの取得
def get_auth_token():
    auth_endpoint = f'{marketo_url}/identity/oauth/token'
    auth_data = {
        'grant_type': 'client_credentials',
        'client_id': client_id,
        'client_secret': client_secret
    }
    response = requests.post(auth_endpoint, data=auth_data)
    return response.json()['access_token']

2-2.ジョブ作成

#（Marketo）ジョブの作成
def create_job(headers):
    endpoint = f'{marketo_url}/bulk/v1/leads/export/create.json'
    body = {
    "fields": [
        ★スマートリストのフィールド★,
    ],
    "format": "CSV",
    "filter": {"smartListId":"★スマートリストID★"},
    }
    response = requests.post(endpoint, headers=headers, data=json.dumps(body))
    job_id = response.json()['result'][0]['exportId']
    return job_id

2-3.ジョブの開始

#（Marketo）ジョブの開始
def start_job(job_id,headers):
    endpoint = f'{marketo_url}/bulk/v1/leads/export/{job_id}/enqueue.json'
    response = requests.post(endpoint, headers=headers)
    return response.json()

2-4.ファイル取得

# (Marketo)ファイル取得
def get_marketo_csv(job_id,headers):
    endpoint = f'{marketo_url}/bulk/v1/leads/export/{job_id}/file.json'
    response = requests.get(endpoint, headers=headers)
    return response

3.関数を実行

def main(request):
  # (Marketo)アクセストークンの取得
  access_token = get_auth_token()
  headers = {
      'Authorization': f'Bearer {access_token}',
      'content-type': 'application/json'
  }
  job_id = create_job(headers)
  start_job(job_id,headers)
  polling_status=polling_job(job_id,headers)
  if polling_status == 'Completed':
      response = get_marketo_csv(job_id,headers)

まとめ

上記にて、無事にスマートリストのデータをCSVで保存することができました。また今回私がやったのは取得するだけではなく、その取得したデータをGCPのBigQueryにテーブルとして保存するということもやっているのでそちらはまた機会のある時に投稿します！

参考

Pythonで マルケトの REST API 使ってLEADデータを取得する - Qiita