https://www.m3tech.blog/entry/2025/03/05/170000

こんにちは、Salesforceのデータを使って推薦システムを作ることになったAI・機械学習チームの農見(@rookzeno)です。Salesforceのデータを扱うエンジニアは、Salesforceのオブジェクトからデータをダウンロードして、Pythonでデータを加工して、他システムにアップロードしたくなる時があると思います。今回はその流れを書いていきます。

refresh tokenを取得する

まずはrefresh tokenを取得出来るようにしましょう。Salesforceのアプリケーションマネージャーから新規接続アプリケーションを作成します。このアプリでOAuth認証を出来るようにします。

次にAPI (OAuth 設定の有効化)のタブでOAuth 設定の有効化をクリックします。すると次のような画面になるので、OAuth 範囲を選択します。データ連携する場合はAPIとrefresh tokenがあれば大丈夫です。コールバックURLにはローカルホスト(http://localhost)を入力します。それと今回は社内での活用で簡易に利用する事を想定しPKCE拡張を外しています。PKCE拡張を利用する場合はcode verifierとのやり取りが必要になるので、token取得方法が変わります。*1

設定したら保存して、接続アプリケーションを作成します。アプリケーションの作成が完了したら、client_idとclient_secretを確認します。アプリケーション一覧から「参照」を押して「コンシューマーキーと秘密の管理」を押してコンシューマの鍵(client_id)とコンシューマーの秘密(client_secret)を確認出来ます。

次はブラウザを使って{ドメインURL(例: https://MyDomainName.my.salesforce.com)}/services/oauth2/authorize?response_type=code&redirect_uri=http%3A%2F%2Flocalhost&client_id={your_client_id}にアクセスします。ドメインURLとyour_client_idを自分のURLとidに変換してください。カスタムドメインを使ってない場合はhttps://login.salesforce.com/にすれば良いです。

そうするとcodeをhttp://localhost/?code=xxxxxxxという形で得られると思うので、このcodeを利用して

curl -v https://MyDomainName.my.salesforce.com/services/oauth2/token  \ 
-d 'grant_type=authorization_code'  \
-d 'code=CODE'  \
-d 'client_id=CLIENT_ID'  \
-d 'client_secret=CLIENT_SECRET'  \ 
-d 'redirect_uri=http://localhost'

を実行することでrefresh tokenを取得することが出来ます。

refresh tokenを使ってデータをダウンロードする

Simple Salesforce

Salesforceでは色々なAPIを提供してくれているのですが、そのPythonラッパーがSimple Salesforceです。今回はこれを使ってSalesforceのオブジェクトからデータをダウンロードします。

refresh tokenを使ってSimple Salesforceの認証をする

Simple Salesforceでは直接refresh tokenを使用した認証はないのですが、access tokenを使った認証は出来るので、refresh tokenからaccess tokenを取得するコードを書いて認証します。

import os
from simple_salesforce import Salesforce

class SalesforceManager:
    def __init__(self):
        self.client_id = os.environ['CLIENT_ID']
        self.client_secret = os.environ['CLIENT_SECRET']
        self.refresh_token = os.environ['REFRESH_TOKEN']
        self.salesforce_url = os.environ['SALESFORCE_URL']
        self.salesforce_token_url = f'{self.salesforce_url}/services/oauth2/token'

    def login_salesforce(self) -> Salesforce:
        session_id = self.get_session_id()
        sf = Salesforce(instance_url=self.salesforce_url, session_id=session_id)
        return sf

    def get_session_id(self) -> str:
        """salesforce_simpleではaccess_tokenをsession_idと呼んでいる"""

        data = {'grant_type': 'refresh_token', 'client_id': self.client_id, 'client_secret': self.client_secret, 'refresh_token': self.refresh_token}
        headers = {'content-type': 'application/x-www-form-urlencoded'}

        response = requests.post(url=self.salesforce_token_url, data=data, headers=headers)
        response.raise_for_status()

        response_data = response.json()
        session_id = response_data.get('access_token')
        return str(session_id)

# これでSimple salesforceのセッションを作成できる
sf = SalesforceManager().login_salesforce()

Simple Salesforceでデータをダウンロードする

基本的にBulk APIを使用してダウンロードするのがおすすめです。普通のAPIだと1回につき数百件しか取ってこれないので、Bulk APIで一気にダウンロードしましょう。取得したデータはlist[dict]という形式です。扱いやすいようにデータフレームに変換しましょう。

import pandas
data = sf.bulk.TMP__c.query("SELECT Id, Name, FROM TMP__c")
df = pd.DataFrame.from_dict(data,orient='columns').drop('attributes',axis=1)

データフレームに出来たら、後はやりたい放題。使いやすい形に加工して、他のシステムにアップロードしましょう。