kintone | Pythonとkintone APIを使ってレコードを追加・更新する

kintone APIを Pythonのスクリプトから呼び出して既存アプリに新規レコードの追加や更新を行います。

スクリプト構成
実行方法

スクリプト構成

以下の3つのスクリプトファイルで構成されています

環境設定保持ファイル・・・config.py
kintone APIとの通信や共通関数をまとめたユーティリティファイル・・・kintone_utils.py
メインスクリプト。CSVデータを読み込み、レコードの追加・更新を行うファイル・・・main.py

環境設定ファイル

kintoneのドメイン、アプリID、APIトークンなどの設定情報が含まれています。また、変数 MODE に基づいて、設定を切り替えることができます。

# config.py

# モードの設定（例：'development', 'production'）
MODE = 'development'

if MODE == 'production':
    KINTONE_DOMAIN = 'xxx0prod.cybozu.com'
    APP_ID = '2'
    API_TOKEN = '********************'
elif MODE == 'development':
    KINTONE_DOMAIN = 'xxx0dev.cybozu.com'
    APP_ID = '4'
    API_TOKEN = '********************'
else:
    raise ValueError(f"Unknown mode: {MODE}")

ユーティリティファイル

kintone APIとの通信や共通関数を提供します。ここでは、レコードのフォーマット、既存レコードの取得、レコードの更新、レコードの追加を行います。

# kintone_utils.py

import requests
import config  # config.pyをインポート

# Kintone APIエンドポイント
record_url = f'https://{config.KINTONE_DOMAIN}/k/v1/record.json'
records_url = f'https://{config.KINTONE_DOMAIN}/k/v1/records.json'

def format_record(row):
    return {
        'SID': {
            'value': str(row['SID'])
        },
        'UserId': {
            'value': str(row['UserId'])
        },
        '名前': {
            'value': str(row['名前'])
        }
    }

def get_existing_records():
    headers = {
        'X-Cybozu-API-Token': config.API_TOKEN
    }
    payload = {
        'app': config.APP_ID,
        'fields': ['SID', 'UserId', '名前', '$id'],
    }

    response = requests.get(records_url, headers=headers, params=payload)

    if response.status_code == 200:
        records = response.json().get('records', [])
        return {record['ID']['value']: record for record in records}
    else:
        print(f"Error retrieving records: {response.text}")
        return {}

def update_record(record_id, updated_record):
    headers = {
        'X-Cybozu-API-Token': config.API_TOKEN,
        'Content-Type': 'application/json'
    }
    payload = {
        'app': config.APP_ID,
        'id': record_id,
        'record': updated_record
    }

    response = requests.put(record_url, headers=headers, json=payload)

    if response.status_code == 200:
        print(f"Record {record_id} updated successfully.")
    else:
        print(f"Error updating record {record_id}: {response.text}")

def add_records(records_to_add):
    headers = {
        'X-Cybozu-API-Token': config.API_TOKEN,
        'Content-Type': 'application/json'
    }
    payload = {
        'app': config.APP_ID,
        'records': records_to_add
    }
    response = requests.post(records_url, headers=headers, json=payload)
    if response.status_code != 200:
        print(f"Error: {response.text}")
    else:
        print(f"Success: {response.json()}")

メインスクリプト

CSVデータを読み込み、既存レコードと比較し、必要に応じてレコードを追加または更新します。

# main.py

import pandas as pd
from kintone_utils import format_record, get_existing_records, update_record, add_records

# CSVファイルの読み込み
csv_file_path = './data/list.csv'  # パスを変更
data = pd.read_csv(csv_file_path)

# レコードを追加または更新する関数
def add_or_update_records_in_kintone(data):
    batch_size = 100  # バッチサイズ（Kintoneの1回のリクエストで送信できる最大レコード数）

    # 既存のIDを取得
    existing_records = get_existing_records()

    records_to_add = []
    for _, row in data.iterrows():
        sid = str(row['SID'])
        record = format_record(row)

        if sid in existing_records:
            existing_record = existing_records[sid]
            updated_record = {k: v for k, v in record.items() if v['value'] != existing_record[k]['value']}
            if updated_record:
                update_record(existing_record['$id']['value'], updated_record)
        else:
            records_to_add.append(record)

            if len(records_to_add) == batch_size:
                add_records(records_to_add)
                records_to_add = []

    # 残りのレコードを送信
    if records_to_add:
        add_records(records_to_add)

# レコードを追加または更新
add_or_update_records_in_kintone(data)

実行方法

main.py を実行して CSVファイルの処理、kintone アプリへレコード追加・更新を開始します。

python main.py