PythonでCSVからSalesforce REST APIに書き込む方法

Posted at 2025-03-07

PythonでSalesforceにデータを連携する方法について記事を書きました。OAuth2認証を使ったREST APIアクセス、CSVからのデータ登録方法を解説しています。Trailhead環境で簡単に試せるので、Salesforceとの連携でお困りの方はぜひご参考に。

この記事の概要

PythonでSalesforceにアクセスする一般的な方法
SalesforceのTrailhead環境(Developer EditionやPlayground)を使った準備
CSVファイルを読み込み、SalesforceのREST APIを使ってデータを書き込むサンプルコード

前提・準備

1. Salesforce環境（Trailhead PlaygroundまたはDeveloper Edition）

TrailheadのPlayground、または開発者向けのDeveloper Editionでテストできます。
設定 → [クイック検索]で「Connected Apps」を検索し、Connected Appの作成が必要です。

2. 組織全体のOAuth設定の確認 (重要)

Salesforce上で Setup(設定) → Identity → OAuth and OpenID Connect Settings を確認します。
以下の項目が有効になっていることを確認してください：
- Allow OAuth Username-Password Flows: オンにする必要があります（パスワードフローを使用するため）

3. Connected Appの作成 (必須)

Salesforce上で Setup(設定) → App Manager → New Connected App をクリックして新規アプリを作成します。
「Enable OAuth Settings(OAuth設定を有効化)」にチェックを入れます。
OAuth and OpenID Connect Settings で以下を設定します：
- Callback URL: http://localhost:8080/callback など（パスワードフローでは実際に使用しませんが必須項目）
- Selected OAuth Scopes: 以下のスコープを選択します
  - 「Access and manage your data (api)」: REST APIアクセスに必要
  - 「Manage user data via APIs (api_integration_user)」: ユーザデータの操作に必要（コンタクト作成など）
  - 「Perform requests at any time (refresh_token, offline_access)」: 長期実行バッチ処理で必要な場合
作成後、取得した Consumer Key (クライアントID) や Consumer Secret (クライアントシークレット) は、Pythonスクリプトでの認証時に使用します。

注意: Summer '23 リリース以降のSalesforce組織では、パスワードフローがデフォルトでブロックされています。スクリプトを実行するには、組織の管理者が「Allow OAuth Username-Password Flows」設定を有効にする必要があります。

4. APIアクセス方式

サンプルコードでは OAuth2 パスワードフロー を使用します：

ユーザ名、パスワード、セキュリティトークン、および Connected App のクライアントID/シークレットを組み合わせた認証方式
一時的なアクセストークンを取得し、そのトークンを使ってAPI操作を行います
注意: この方式はSummer '23リリース以降は組織の設定で明示的に許可する必要があります

手順

1. Salesforceの認証情報を取得

セキュリティトークンの取得:
- Setup(設定) → Users(ユーザ) → Users(ユーザ) → 自分のユーザ名 → Reset My Security Token(セキュリティトークンのリセット)
Connected Appの情報
- Connected App から Consumer Key と Consumer Secret を取得

2. CSVファイルの準備

FirstName,LastName,Email
Taro,Yamada,taro.yamada@example.com
Hanako,Suzuki,hanako.suzuki@example.com
John,Smith,john.smith@example.com

3. Pythonサンプルコード

import csv
import requests

# 設定
USERNAME = 'your_username@example.com'
PASSWORD = 'your_salesforce_password'
SECURITY_TOKEN = 'XXXXXXXXXX'
LOGIN_URL = 'https://login.salesforce.com/services/oauth2/token'
CLIENT_ID = 'Connected_App_Consumer_Key'
CLIENT_SECRET = 'Connected_App_Consumer_Secret'
API_VERSION = 'v57.0'
CSV_FILE_PATH = 'contacts.csv'
SOBJECT_NAME = 'Contact'

def get_access_token():
    data = {
        'grant_type': 'password',
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
        'username': USERNAME,
        'password': PASSWORD + SECURITY_TOKEN
    }
    response = requests.post(LOGIN_URL, data=data)
    if response.status_code != 200:
        raise Exception(f"Failed to get token: {response.text}")
    return response.json()['access_token'], response.json()['instance_url']

def create_contact_records(access_token, instance_url, records):
    headers = {
        'Authorization': f'Bearer {access_token}',
        'Content-Type': 'application/json'
    }
    created_ids = []
    for record in records:
        url = f"{instance_url}/services/data/{API_VERSION}/sobjects/{SOBJECT_NAME}/"
        response = requests.post(url, headers=headers, json=record)
        if response.status_code == 201:
            created_ids.append(response.json().get('id'))
        else:
            print("Error creating record:", response.text)
    return created_ids

def main():
    access_token, instance_url = get_access_token()
    records_to_insert = []
    with open(CSV_FILE_PATH, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            contact_record = {
                'FirstName': row['FirstName'],
                'LastName': row['LastName'],
                'Email': row['Email']
            }
            records_to_insert.append(contact_record)
    created_ids = create_contact_records(access_token, instance_url, records_to_insert)
    print("Created Record IDs:", created_ids)

if __name__ == '__main__':
    main()

4. 認証フローの解説

このコードで使用しているのは OAuth2 パスワードフロー です：

ユーザ名、パスワード、セキュリティトークン、Connected AppのクライアントID/シークレットを使って認証
login.salesforce.com に対して認証リクエストを送信
レスポンスからアクセストークンとインスタンスURLを取得
取得したアクセストークンを使ってREST API操作を実行

5. 実行方法

contacts.csv を作成し、同じフォルダにPythonスクリプトを保存。
USERNAME、PASSWORD、SECURITY_TOKEN、CLIENT_ID、CLIENT_SECRET を正しく設定。
以下のコマンドを実行:
```
python script.py
```
Salesforceの画面で Contact にデータが登録されたことを確認。

トラブルシューティング

認証エラー: ユーザ名やパスワード、セキュリティトークン、クライアントID/シークレットが正しいか確認。
APIバージョンエラー: API_VERSION を最新または適切なバージョンに変更。
マッピングエラー: CSVのカラム名とSalesforceのフィールド名が一致しているか確認。
アクセス権限: 使用しているユーザーにAPIアクセス権限が付与されているか確認。

効率化のためのヒント

一括処理: 多数のレコードを処理する場合は、Composite API や Bulk API の使用を検討。
エラーハンドリング: 本番環境では、より堅牢なエラー処理とリトライロジックを実装。
サンドボックス環境: 本番環境への接続前に、サンドボックス環境でテスト（login.salesforce.com の代わりに test.salesforce.com を使用）。

まとめ

Pythonを使ってCSVデータをSalesforce REST API経由で登録する方法を紹介。
OAuth2 パスワードフローを使用した認証方法を実装。
Trailhead PlaygroundやDeveloper Editionで簡単にテスト可能。
小規模データならREST APIでOK、大規模データならBulk APIの検討を推奨。

この方法を活用してSalesforceのデータ連携を効率化しましょう！

マインドマップ

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up