Pythonで実現する業務自動化カスタム項目一括作成【前編】 - TerraSkyBase | テラスカイを支える人とテクノロジーの情報を発信する基地局

はじめに

Salesforceでオブジェクトのカスタマイズをしていると、「またカスタム項目を追加しないといけないのか…」という場面がありませんか。
項目が多い要件だと、同じような操作を何十回も繰り返すことになり、「これ、なぜ一括でできないの？」と叫びたくなりますよね。
XMLファイルを使ってSFDXやMetadata APIで一括反映する方法もありますが、「XMLの構造が難しい」「ツールの操作が煩雑」と感じる方も多いのではないでしょうか。

本記事ではより作業効率を上げるために、Pythonを使用しSalesforceのカスタム項目を一括で作成するアプリを作成します。

機能概要

このアプリは、Pythonの simple-salesforce ライブラリを使用して、SalesforceのSandboxまたはDeveloper環境へログインし、CSVファイルを読み込むことでカスタム項目を一括作成することができます。

ユーザーインターフェースはFletを用いて構築しており、Webアプリのような画面上で操作が可能です。これにより、コマンドラインに不慣れな方でも直感的に操作できます。

また、CSVファイルの入力形式に迷わないように、テンプレートCSVをダウンロードできる機能も用意しており、誰でもスムーズに利用を開始できます。

完成イメージ

1. ログイン画面

2. CSVアップロード画面

前提条件

・VSCodeにてPythonの実行環境が構築されていること
　（Pythonパッケージのインストールや、仮想環境の作成ができる状態）

実行環境

・Python　3.11.4
・simple-salesforce　1.12.6
・Flet　0.27.5

ライブラリのインストール

まず、ライブラリのインストールを行います。
以下に記載するコマンドをコマンドプロンプトへ入力しインストールしてください。

1.simple-salesforce をインストール

PythonとSalesforceを連携させるため、simple-salesforceを利用します。

pip install simple-salesforce

simple-salesforce

2.Flet をインストール

今回はGUIにFletを使用します。
Pythonのみで簡単に実装ができ、モダンUIでの開発が可能なためFletを採用しました。

pip install flet

Flet

カスタム項目の一括作成アプリの作成

1. ディレクトリ構成

以下のディレクトリ構成とします。

root.
| ---- Salesforce Automation Tool
    | ---- log
    | ---- sf_main.py
    | ---- sf_shared_state.py
    | ---- sf_login_view.py
    | ---- sf_csv_upload_view.py
    | ---- sf_create_fields.py

ディレクトリ構成

2. エントリーポイントの作成（sf_main.py）

アプリ起動時にログイン画面を最初に表示するためのエントリーポイントを作成します。
以下のコードを、sf_main.pyに記述してください。

import flet as ft
import sf_login_view

login = sf_login_view.login

def main(page: ft.Page):
    page.views.append(login(page))
    page.update()


if __name__ == '__main__':
    ft.app(target=main, view=ft.FLET_APP)

sf_main.py

3. アプリ全体で共有する変数の定義（sf_shared_state.py）

各画面で共有する変数を定義します。
以下のコードを、sf_shared_state.pyに記述してください。

SF_MDAPI = None        # Salesforce Metadata API 接続用のオブジェクト（ログイン後にセット）
BG_COLOR = '#fffafa'        # 背景色
PROGRESS_TEXT = "項目作成中です..."        # 処理中に表示するメッセージ

sf_shared_state.py

4. ログイン画面（sf_login_view.py）

ログイン画面を作成します。sf_main.pyから呼び出されるプログラムです。
以下のコードを、sf_login_view.pyに記述してください。

import flet as ft
from simple_salesforce import Salesforce, SalesforceAuthenticationFailed
from sf_csv_upload_view import csv_upload_view
import sf_shared_state

# ログインする際の情報
CREDENTIALS_DICT = {
    "DOMAIN" : '',             # 環境
    "USER_NAME" : '',          # ユーザー名
    "PASSWORD" : '',           # パスワード
    "SECURITY_TOKEN" : ''      # セキュリティトークン
}

ENV_SANDBOX = 'Sandbox'                         # 接続先環境 Sandbox
ENV_DEVELOPER_EDITION = 'Developer Edition'     # 接続先環境 Developer Edition

LABEL_COLOR = ft.TextStyle(color="black")
TEXT_COLOR = ft.TextStyle(color="black")


def login(page: ft.Page):

    # アプリタイトル   
    page.title = "Salesforce Automation Tool"
    
    # ウィンドウサイズを設定
    page.window.width = 640
    page.window.height = 480
    page.window.full_screen = False
    page.window.resizable = False
    page.window.maximizable = False
    
    # 入力値の設定
    def on_field_blur(e):
        key = e.control.data["key"]
        CREDENTIALS_DICT[key] = e.control.value

    # 選択した環境のドメイン設定
    def on_field_change(e):
        key = e.control.data["key"]
        domain = e.control.value
        
        if domain ==  ENV_DEVELOPER_EDITION:
            CREDENTIALS_DICT[key] = 'login'
        elif domain == ENV_SANDBOX:
            CREDENTIALS_DICT[key] = 'test'

    # ログインボタンのクリックイベント
    def on_button_click(e): 
        is_succses, sf = sf_login()
        
        if not is_succses:
            page.open(ft.SnackBar(ft.Text("Salesforceへのログインに失敗しました。")))
        else :
            sf_shared_state.SF_MDAPI = sf.mdapi
            
            #アップロード画面へ遷移
            page.views.append(csv_upload_view(page))    
        
        page.update()
    
    # 入力欄を作成
    user_name_field = ft.TextField(
        label="ユーザー名", 
        width=300, 
        on_blur=on_field_blur, 
        data={"key": "USER_NAME"}, 
        label_style=LABEL_COLOR, 
        text_style=TEXT_COLOR
    )
    # パスワード欄を作成
    password_field = ft.TextField(
        label="パスワード",
        width=300,
        on_blur=on_field_blur, 
        data={"key": "PASSWORD"}, 
        password=True,
        can_reveal_password=True,
        label_style=LABEL_COLOR,
        text_style=TEXT_COLOR
    )
    # セキュリティトークン欄を作成
    security_token_field = ft.TextField(
        label="セキュリティトークン",
        width=300,
        on_blur=on_field_blur,
        data={"key": "SECURITY_TOKEN"},
        password=True,
        can_reveal_password=True,
        label_style=LABEL_COLOR,
        text_style=TEXT_COLOR
    )
    
    # 環境選択欄を作成
    env_field = ft.Dropdown(
        label="接続先環境",
        width=300,
        on_change=on_field_change,
        data={"key": "DOMAIN"},
        options=[
            ft.dropdown.Option(ENV_SANDBOX),
            ft.dropdown.Option(ENV_DEVELOPER_EDITION),
        ],
        label_style=LABEL_COLOR,
        text_style=TEXT_COLOR
    )

    # ボタンを作成
    login_button = ft.ElevatedButton(
        "ログイン", 
        on_click=on_button_click,
        width=300,
        height=60,
        style=ft.ButtonStyle(
            shape=ft.RoundedRectangleBorder(radius=10),
            text_style=ft.TextStyle(size=20)
        )
    )

    # 縦に中央揃えで配置
    column = ft.Column(
        [env_field, user_name_field, password_field, security_token_field, login_button],
        alignment=ft.MainAxisAlignment.CENTER,
        spacing=20
    )
    
    return ft.View(
        "/login",
        [
            ft.Container(
                content=column,
                alignment=ft.alignment.center,
                expand=True
            )
        ],
        bgcolor = sf_shared_state.BG_COLOR
    )

# Salesforceにログイン
def sf_login():
    try:
        sf = Salesforce(
            username=CREDENTIALS_DICT["USER_NAME"],
            password=CREDENTIALS_DICT["PASSWORD"],
            security_token=CREDENTIALS_DICT["SECURITY_TOKEN"],
            domain=CREDENTIALS_DICT["DOMAIN"]
        )
        return True, sf
    except SalesforceAuthenticationFailed:
        return False, None
    except Exception as e:
        return False, None

sf_login_view.py

5. CSVアップロード画面（sf_csv_upload_view）

CSVファイルをアップロードしたり、CSVファイルのテンプレートをダウンロードできる画面です。
ログイン画面の「ログイン」ボタンをクリックし、ログインできた際に表示されます。
以下のコードを、sf_csv_upload_view.pyに記述してください。

import flet as ft
import os
import sys
from datetime import datetime
import csv
from sf_create_fields import create_fields
import sf_shared_state

BASE_FILE_NAME = "csv_template.csv" # テンプレートCSVの名前

selected_file_path = ft.Ref[str]()


def csv_upload_view(page: ft.Page):
    
    # CSVファイルの選択
    def on_file_selected(e: ft.FilePickerResultEvent):
        if e.files:
            file = e.files[0]
            if file.name.endswith(".csv"):
                selected_file_path.current = file.path
                selected_file.value = file.name
                selected_file.update()
            else :
                selected_file.value = "CSVファイルのみアップロードできます"
                selected_file_path.current = None
                selected_file.update()
    
    file_picker = ft.FilePicker(on_result=on_file_selected)
    page.overlay.append(file_picker)
    
    # CSVテンプレートをダウンロードボタンのクリックイベント
    def click_download_template(e):
        folder_picker = ft.FilePicker()
        
        def create_template(save_path):
            
            # ヘッダー
            headers = [
                'オブジェクト名',
                '項目名',
                '表示ラベル',
                'データ型',
                '文字数',
                '小数点の位置',
                '値'
            ]
            
            full_path = os.path.join(save_path , BASE_FILE_NAME)
            
            # ファイル名が重複している場合に連番とする
            counter = 1
            filename, ext = os.path.splitext(BASE_FILE_NAME)
            while os.path.exists(full_path):
                full_path = os.path.join(save_path, f"{filename}({counter}){ext}")
                counter += 1

            # CSVファイルを書き出す
            with open(full_path, mode='w', newline='', encoding='shift_jis') as file:
                writer = csv.writer(file)
                writer.writerow(headers)

        def folder_result(e: ft.FilePickerResultEvent):
            if e.path:                
                create_template(e.path)
                dlg = ft.AlertDialog(
                    content=ft.Text("ダウンロードしました", text_align=ft.TextAlign.CENTER, size=24),
                )
                page.open(dlg)

        folder_picker.on_result = folder_result
        page.overlay.append(folder_picker)
        page.update()
        folder_picker.get_directory_path(dialog_title="保存先フォルダを選んでください")
    
    # CSVテンプレートをダウンロードを作成
    download_file_button = ft.ElevatedButton(
        "CSVテンプレートをダウンロード",
        on_click=click_download_template,
        width=160, 
        height=30,
        style=ft.ButtonStyle(
            shape=ft.RoundedRectangleBorder(radius=10),
            text_style=ft.TextStyle(size=12)
        )
    )
    
    # ファイル名を表示するためのテキストを作成
    selected_file = ft.Text(
        "ファイルが選択されていません",
        color="black",
        max_lines=2,
        overflow=ft.TextOverflow.ELLIPSIS,
        style=ft.TextStyle(size=14)
    )
    
    # CSVを選択ボタンを作成
    pick_file_button = ft.ElevatedButton(
        "CSVを選択", 
        on_click=lambda _: file_picker.pick_files(allow_multiple=False, allowed_extensions=["csv"]),
        width=120, 
        height=60,
        style=ft.ButtonStyle(
            shape=ft.RoundedRectangleBorder(radius=10),
            text_style=ft.TextStyle(size=20)
        )
    )
    
    
    # ダイアログ内に表示するテキストを作成
    progress_text =  ft.Text("項目作成中です...")
    
    # ダイアログ内に表示するボタンを作成
    dlg_btn = ft.TextButton(
                    "OK", 
                    on_click=lambda e: page.close(dlg_modal),
                    disabled=True
                )
    
    # 項目作成中のダイアログを作成
    dlg_modal = ft.AlertDialog(
        modal=True,
        content=ft.Column(
            [progress_text],
            tight=True,
        ),  
        actions=[dlg_btn],
        actions_alignment=ft.MainAxisAlignment.CENTER,
    )
    
    # 項目作成開始ボタンのクリックイベント
    def upload_file(e):
        if not selected_file.value.startswith("ファイルが選択されていません"):
            page.open(dlg_modal)
            is_completed, traceback_text = create_fields()
            if is_completed:
                    progress_text.value = "項目作成完了!"
            else :
                progress_text.value = "エラーが発生しました。\n詳細はエラーログ（logフォルダ内のファイル）をご確認ください。"
                create_error_log(traceback_text)
            dlg_btn.disabled = False
            page.update()
            
        else :
            print("ファイルが選択されていません")
    
    # 項目作成開始ボタンを作成
    upload_button = ft.ElevatedButton(
        "項目作成開始", 
        on_click=upload_file,
        width=180, 
        height=60,
        style=ft.ButtonStyle(
            shape=ft.RoundedRectangleBorder(radius=10),
            text_style=ft.TextStyle(size=20)
        )
    )
    
    download_button_row = ft.Column(
        [download_file_button], 
        alignment=ft.CrossAxisAlignment.CENTER
    )

    pick_file_row = ft.Column(
        [pick_file_button, selected_file],
        horizontal_alignment = ft.CrossAxisAlignment.CENTER
    )
    
    pick_file_row_container = ft.Container(
        content=pick_file_row,
        alignment=ft.alignment.center,
        padding=ft.padding.only(top=10, bottom=20),
        border=ft.border.all(2, ft.colors.BLUE),  # 幅2px、青色の枠線
        border_radius=10,  # 角丸
        width=480,
        height=140
    )

    upload_button_row = ft.Column(
        [upload_button], 
        alignment=ft.CrossAxisAlignment.CENTER,
    )
    
    upload_button_row_container = ft.Container(
        content=upload_button_row,
        alignment=ft.alignment.center
    )

    column = ft.Column(
        [download_button_row, pick_file_row_container, upload_button_row_container],
        horizontal_alignment = ft.CrossAxisAlignment.CENTER,
        alignment=ft.MainAxisAlignment.SPACE_EVENLY
    )
    
    file_upload_Container = ft.Container(
        content=column,
        expand=True,
        margin=ft.margin.only(top=30)
    )
    
    return ft.View(
        "/upload",
        [file_upload_Container],
        bgcolor=sf_shared_state.BG_COLOR
    )

# 項目作成のエラーログを出力する
def create_error_log(error_text):
    
    if getattr(sys, 'frozen', False):
        base_path = os.path.dirname(sys.executable)
    else:
        # 通常のPython実行時
        base_path = os.path.dirname(os.path.abspath(__file__))
        
    # ログフォルダ作成（なければ）
    log_dir = os.path.join(base_path, "log")
    os.makedirs(log_dir, exist_ok=True)

    # ログファイルのファイル名生成
    now = datetime.now()
    formatted = now.strftime("%Y-%m-%d-%H-%M-%S")
    file_path = os.path.join(log_dir, f"error_{formatted}.log")

    # ログファイルに書き込み
    with open(file_path, mode="w", encoding='utf-8') as f:
        f.write(error_text)

sf_csv_upload_view.py

6. CSVファイルから項目を作成する（sf_create_fields.py）

CSVファイルを読み込み、項目を作成するプログラムです。
今回は使用頻度の多い、テキスト型、数値型、選択リスト型、日付型のみ作成対象となります。
以下のコードを、sf_create_fields.pyに記述してください

import sf_csv_upload_view as csv_view
import csv
import sf_shared_state
import sys

# csvヘッダーの読み込み順を固定
props_order = ["データ型","オブジェクト名","項目名","表示ラベル","文字数","小数点の位置","値"]

# CSVヘッダーをキーとした共通項目を設定
salesforce_field_common_props = {
    "オブジェクト名" : "fullName",
    "項目名" : "fieldName",
    "表示ラベル" : "label",
    "データ型" : [
        {"テキスト":"Text"},
        {"数値":"Number"},
        {"選択リスト":"Picklist"},
        {"日付":"Date"},
    ]
}

# 共通プロパティを設定する
def mappinng_common_props(key, value):
    
    if key in salesforce_field_common_props:
            
            field_key = salesforce_field_common_props[key]
            
            if key == "データ型" :
                types = salesforce_field_common_props[key]
                prop_value = next((v for d in types for k, v in d.items() if k == value), None)
                
                if prop_value != None:
                    return 'type', sf_shared_state.SF_MDAPI.FieldType(prop_value)
                else : return 'type', None
                
            else :
                return field_key, value
            
    else : return None, None

# 必須項目をデータ型ごとに設定する
def mapping_required_props(field, key, value):

    # テキスト
    if field["type"] == "Text":
        if key == "文字数":
            field["length"] = value
            
    # 数値
    elif field["type"] == "Number":
        if key == "文字数":
            field["precision"] = value
            
        if key == "小数点の位置":
            field["scale"] = value

    # 選択リスト
    elif field["type"] == "Picklist":
        
        def get_value_struct(v):
            return {
                        "fullName" : v,
                        "default" : False,
                        "label" : v
            }
        if key == "値":
            value_list = str(value).split(";")
            
            field["valueSet"] = {
                    "valueSetDefinition" : {
                        "sorted" : False,
                        "value" : []
                    }
            }
            
            for v in value_list:
                field["valueSet"]["valueSetDefinition"]["value"].append(get_value_struct(v))

# 項目の作成
def create_fields():
    file_path = csv_view.selected_file_path.current
    
    # CSVヘッダーを並び替える
    def sort_row(row):
        sorted_dict = {}
        for key in props_order:
                if key in row:
                    sorted_dict[key] = row[key]
        return sorted_dict

    # カスタム項目のMetadataを設定
    def get_custom_field_props():
        return {
                    "fullName" : "",
                    "label" : "",
                    "type" : ""
                }
    
    # CSVファイルを読み込む
    with open(file_path, encoding='shift_jis') as f:
        reader = csv.DictReader(f)
        custom_fields = []
        
        add_count = 0
        
        for i, row in enumerate(reader):
            
            sorted_dict = sort_row(row)
            
            custom_fields_props = get_custom_field_props()
            
            error_flg = False
            
            for key, value in sorted_dict.items():
                
                value = value if value is not None else ''
                prop_key, prop_value = mappinng_common_props(key, value)
                
                # salesforce_field_common_propsで指定したデータ型以外は処理しない
                if prop_key == 'type' and prop_value == None :
                    error_flg = True
                
                if not error_flg :
                    
                    # 各プロパティに値を設定
                    if prop_key != None or prop_value != None: 
                        
                        # fullNameはオブジェクト名.項目名とするため、文字列結合する
                        if prop_key == "fieldName":
                            custom_fields_props["fullName"] = f'{custom_fields_props["fullName"]}.{prop_value}'
                        else :
                            custom_fields_props[prop_key] =  prop_value

                    mapping_required_props(custom_fields_props, key, value)
                    
            if not error_flg :
                
                custom_field = sf_shared_state.SF_MDAPI.CustomField(**custom_fields_props)
                custom_fields.append(custom_field)
                add_count += 1

        # 一度に10個しか作成できないため、10項目ごとに作成する 
        grouped =  [custom_fields[i:i+10] for i in range(0, len(custom_fields), 10)]
        
        error_value = ''
        for fields in grouped :
            try:
                sf_shared_state.SF_MDAPI.CustomField.create(fields)
            except Exception as e:
                exc_type, exc_value, exc_traceback = sys.exc_info()
                error_value += '\n' + str(exc_value)
                
    return not bool(str(error_value)) , error_value

sf_create_fields.py

使い方

1. ログイン画面

まず、VSCodeでsf_main.pyを開き実行します。

以下の項目を入力し「ログイン」ボタンをクリックします。
・接続先環境：項目を作成したい環境を選択します。
・ユーザー名：項目を作成するユーザー名を入力します。
・パスワード：パスワードを入力します。
・セキュリティトークン：ログインするユーザーのセキュリティトークンを入力します。

ログインに成功するとCSVアップロード画面へ遷移します。
失敗した場合は、画面下部に【Salesforceへのログインに失敗しました。】と表示されます。

セキュリティトークンが不明な場合は、個人設定から「私のセキュリティトークンのリセット」 →「セキュリティトークンのリセット」ボタンをクリックすることで、ユーザーのメールアドレスにセキュリティトークンが記載されたメールが送信されます。

2. CSVアップロード画面

CSVアップロード画面では、「CSVテンプレートをダウンロード」ボタンをクリックし、テンプレートファイルを保存してください。

ダウンロードしたCSVファイルに作成する項目を入力します。
必要なプロパティは以下の表の通りです。

列名	説明	入力例
オブジェクト名	作成対象のオブジェクト名（必須）	Account
項目名	作成するカスタム項目のAPI名（必須）	CustomText__c
表示ラベル	項目の表示名（必須）	カスタムテキスト
データ型	データ型（必須）	テキスト
文字数	テキスト型や数値型の場合の最大文字数（テキスト型、数値型の場合のみ入力）	255
小数点の位置	数値型の場合の小数点以下の桁数（数値型の場合のみ入力）	0
値	選択リストの値（セミコロン区切りで入力）	赤;青;黄

以下は入力後のCSVファイルです。

CSVファイルが完成したら「CSVを選択」ボタンで、編集したCSVファイルを選んでください。
画面にファイル名が表示されたことを確認します。
「項目作成開始」ボタンを押すと、Salesforceへの項目作成処理が始まります。
処理中はダイアログ内に「項目作成中です...」と表示され、完了すると「項目作成完了!」と表示されます。
「OK」ボタンでダイアログを閉じます。

エラーが発生した場合は画面にエラーメッセージが表示されます。
詳細は、アプリのlogフォルダ内に自動生成されるエラーログを確認してください。
エラーでない項目のみ作成されますのでご注意ください。

ご利用時の注意点

1. Salesforce側で必要な権限について
Salesforceにカスタム項目を作成するためには、ログイン時のユーザーに以下の権限が必要です。
・API の有効化
・メタデータ API 関数を使用したメタデータの変更
・アプリケーションのカスタマイズ

2. API要求数について
・ログイン時と項目作成時にAPIを呼び出します。API 要求数が上限に達すると利用できなくなります。

3. 入力情報に誤りがないか確認する
・ユーザー名、パスワード、セキュリティトークンはいずれも正確に入力してください。

4. CSVファイルの形式・文字コードに注意
・CSVファイルは Shift-JIS 形式で保存してください。

5. 項目名やデータ型の記述に注意
・項目名（API名）はSalesforceの命名規則に従って記述してください。
・データ型の記述ミス（例: Text や Number など）はエラーの原因になります。正確に入力してください。

6. 項目レベルセキュリティ
・作成時は項目レベルセキュリティが設定されていません。忘れずに設定しましょう。

7. ページレイアウト
・項目を作成しただけではページレイアウトには自動で表示されません。必要な項目を手動で追加してください。

最後に

今回は、Salesforceのカスタム項目作成を効率化するためのアプリについてご紹介しました。

手動での項目作成はどうしても手間がかかり、ミスも発生しやすい作業です。このアプリを活用することで、CSVファイルを使った一括設定が可能になり、作業の大幅な効率化が期待できます。

ただし、Salesforceの仕組み上、API使用制限や権限設定、CSVフォーマットの正確性など、事前に注意すべき点もいくつかあります。本記事内でご紹介したポイントを参考に、安全にご利用ください。

後編では、このアプリを一度作れば他のメンバーにもそのまま配布して利用してもらえるように、デスクトップアプリとしてパッケージ化する方法をご紹介します。

最後までご覧いただきありがとうございました。

はじめに

機能概要

完成イメージ

1. ログイン画面

2. CSVアップロード画面

前提条件

実行環境

ライブラリのインストール

1.simple-salesforce をインストール

2.Flet をインストール

カスタム項目の一括作成アプリの作成

1. ディレクトリ構成

2. エントリーポイントの作成 （sf_main.py）

3. アプリ全体で共有する変数の定義（sf_shared_state.py）

4. ログイン画面（sf_login_view.py）

5. CSVアップロード画面（sf_csv_upload_view）

6. CSVファイルから項目を作成する（sf_create_fields.py）

使い方

1. ログイン画面

2. CSVアップロード画面

ご利用時の注意点

最後に

2. エントリーポイントの作成（sf_main.py）