quickbooks-online-api

QuickBooks Online API エキスパートガイド

概要

QuickBooks Online API を使用すると、QuickBooks Online の企業の会計データや業務に包括的にアクセスできます。このスキルを利用すると、請求書の発行、支払い、顧客管理、在庫追跡、財務レポートに対応する連携を構築できます。API は認証に OAuth 2.0 を使用し、顧客、請求書、支払い、商品、勘定科目など、主要なすべての会計エンティティに対する操作をサポートしています。

QuickBooks Online API は REST ベースで、JSON または XML 形式のレスポンスを返し、Java、Python、PHP、Node.js、C# 向けの SDK を提供しています。サンドボックス（開発用）環境と本番環境の両方をサポートしています。

このスキルを使用する場面

次のような場合にこのスキルを使用します。

• 会計業務の自動化に向けて QuickBooks との連携を構築する
• 請求書発行のワークフローや支払い処理を実装する
• 顧客または仕入先の管理機能を作成する
• QuickBooks Online API の認証（OAuth2）を扱う
• API エラーや検証失敗のトラブルシューティングを行う
• 大量のデータ更新に向けてバッチ操作を実装する
• データ同期のために変更データキャプチャ（CDC）や Webhook を設定する
• 複数通貨や国際会計に対応する連携を設計する
• QuickBooks のデータを活用したレポートや分析機能を構築する
• QuickBooks Online との間でデータを移行する

会計連携の設計

会計同期や仕様をレビューする際は、API 呼び出しを選ぶ前に、既存の文書と支払いのライフサイクルをたどってください。以下の要件は、後述する簡略化された API の例に適用されます。これらの例は、完全な消し込み実装ではありません。

• 所有者を特定する： ローカルの請求書／仕入先請求書、その業務上の所有者、それを公開したり決済を変更したりする権限を持つシステムを特定します。別のプロバイダーが、同じ債務をすでに QuickBooks に公開している場合があります。既存のレコードと照合してから追加し、すべての外部仕入先請求書がローカルの仕入先請求書と一対一で対応すると想定しないでください。
• 売掛金と買掛金を分ける： 顧客からの入金には Payment を、仕入先への決済には BillPayment を使用します。支払いと文書の割り当て、一部決済、未消込の顧客入金を保持します。残高はローカルの支払い台帳から算出してください。支払い済みフラグを繰り返し設定したり、支払済み金額を加算したりしても、消し込みにはなりません。
• 転記の境界を定義する： 編集可能なローカルの下書きと、発行済み請求書または承認済み買掛金を区別します。エクスポートのイベントを明示的に選択してください。会計エクスポートによって、顧客へのメールが二重に送信されたり、銀行決済が開始されたりしてはなりません。
• 金額の意味を保持する： 保存済みの文書明細と正確な金額をエクスポートします。品目／収益／費用のマッピング、税金、通貨、丸め、手数料、クレジット、総額と純額のどちらで決済するかを定義します。勘定科目表全体の同期を対象外としても、取引の転記に必要なマッピングはなくなりません。
• 取引の識別情報を保持する： マッピングは、接続、環境、realm、エンティティ種別、ローカル ID、リモート ID ごとに範囲を限定します。顧客／仕入先だけでなく、請求書、仕入先請求書、支払い、割り当ての識別情報も含めます。ローカルで無効化した後もマッピングを保持してください。再発行は新しい文書です。別の会社に再接続しても、以前のマッピングを再利用できません。
• 復旧可能な配信にする： エクスポート作業は業務トランザクションとともに永続化し、既存のキュー／アウトボックスを通じて処理します。成功後にリモート ID を保存するだけでは、作成応答が失われた場合に対応できません。サポートされている安定したリクエスト ID と、結果が不確定な場合の照合を使用します。更新には SyncToken を使用し、会計上の編集内容が競合した場合の再試行前に、フィールドの所有権に関する方針を明示的に適用します。
• 訂正に対応する： 無効化、クレジット、返金、削除／再割り当てされた支払い、再実行時の動作を定義します。未対応のケースは、照合上の例外として可視化し、文書を暗黙に支払い済みと判断しないでください。ポーリングのチェックポイントは、処理が永続化された後にのみ進めます。Webhook を受信しても、ポーリング期間を飛ばしてはなりません。
• 接続を保護する： 暗号化された認証情報のストレージ、テナント制約と RLS、ユーザー／組織に紐付けられた有効期限付き OAuth 状態、シリアル化されたリフレッシュトークンのローテーションを再利用します。第二の連携フレームワークを追加せず、リポジトリの既存パターンを再利用してください。

Laneweaver TMS の作業では、スキーマ、同期トリガー、支払いのインポート、または QuickBooks の issue #146 への変更を提案する前に、 Laneweaver A/P and A/R design を読んでください。同リポジトリ固有のルールは、他の製品に自動的に適用されるものではありません。

認証と OAuth2 の設定

OAuth 2.0 フロー

QuickBooks Online API では、OAuth 2.0 認証が必要です。このフローは以下の手順で構成されます。

1. developer.intuit.com で アプリを登録 し、Client ID と Client Secret を取得します
2. ユーザーを認証 URL に 誘導 し、QuickBooks の会社へのアクセスを許可してもらいます
3. 認証コードをトークンに交換 します（アクセストークン＋リフレッシュトークン）
4. API リクエストの Authorization: Bearer ヘッダー でアクセストークンを使用します
5. アクセスを維持するため、有効期限が切れる前にトークンを更新 します

トークンのライフサイクル

アクセストークン：

• **3600 秒（1 時間）**有効です
• Authorization ヘッダーに含めます：Authorization: Bearer {access_token}
• 期限切れの場合、401 Unauthorized を返します

リフレッシュトークン：

• 発行から100 日間有効です
• 新しいアクセストークンとリフレッシュトークンのペアを取得するために使用します
• 以前のリフレッシュトークンは、新しいトークンの発行から 24 時間後に期限切れになります
• 常に最新のリフレッシュトークンを使用してください

トークンの更新パターン

Node.js の例：

const oauthClient = require('intuit-oauth');

// Refresh access token
oauthClient.refresh()
  .then(function(authResponse) {
    const newAccessToken = authResponse.token.access_token;
    const newRefreshToken = authResponse.token.refresh_token;
    const expiresIn = authResponse.token.expires_in; // 3600 seconds

    // Store new tokens securely (database, encrypted storage)
    console.log('Tokens refreshed successfully');
  })
  .catch(function(e) {
    console.error('Token refresh failed:', e.originalMessage);
    // Handle re-authentication if refresh token is invalid
  });

Python の例：

from intuitlib.client import AuthClient

auth_client = AuthClient(
    client_id='YOUR_CLIENT_ID',
    client_secret='YOUR_CLIENT_SECRET',
    redirect_uri='YOUR_REDIRECT_URI',
    environment='sandbox'  # or 'production'
)

# Refresh tokens
auth_client.refresh(refresh_token='STORED_REFRESH_TOKEN')

# Get new tokens
new_access_token = auth_client.access_token
new_refresh_token = auth_client.refresh_token

ベストプラクティス

• 事前に更新する：トークンが期限切れになる前に更新してください（例：50 分経過後）
• 安全に保存する：データベース内のトークンを暗号化し、バージョン管理にコミットしないでください
• 401 レスポンスを処理する：認証エラー時にトークンの更新を自動的に試みてください
• Realm ID（Company ID）：OAuth 中に返された realmId を保存してください。すべての API 呼び出しで必要です
• スコープ：必要なスコープ（accounting、payments など）のみをリクエストしてください

コアエンティティのリファレンス

顧客

QuickBooks の顧客およびサブ顧客（ジョブ）を表します。

主なフィールド：

• Id（string、読み取り専用）：一意の識別子
• DisplayName（string、必須）：顧客の表示名（一意である必要があります）
• GivenName、FamilyName（string）：名と姓
• CompanyName（string）：法人顧客の会社名
• PrimaryEmailAddr（object）：メールアドレス { "Address": "[email protected]" }
• PrimaryPhone（object）：電話番号 { "FreeFormNumber": "(555) 123-4567" }
• BillAddr、ShipAddr（object）：請求先住所と配送先住所
• Balance（decimal、読み取り専用）：現在の未払い残高
• Active（boolean）：顧客が有効かどうか
• SyncToken（string、更新時に必須）：オプティミスティックロック用のバージョン番号

参照型：取引では CustomerRef を使用します：{ "value": "123", "name": "Customer Name" }

請求書

顧客に送付する販売請求書を表します。

主なフィールド：

• Id（string、読み取り専用）：一意の識別子
• DocNumber（string）：請求書番号（指定されていない場合は自動生成）
• TxnDate（date）：取引日（YYYY-MM-DD 形式）
• DueDate（date）：支払期日
• CustomerRef（object、必須）：顧客への参照 { "value": "customerId" }
• Line（array、必須）：請求書の明細項目（「明細項目」セクションを参照）
• TotalAmt（decimal、読み取り専用）：計算された合計金額
• Balance（decimal、読み取り専用）：未払い残高
• EmailStatus（enum）：NotSet、NeedToSend、EmailSent
• BillEmail（object）：請求書の送付先となる顧客のメールアドレス
• TxnTaxDetail（object）：税額計算の詳細
• LinkedTxn（array）：関連する取引（支払い、クレジットメモ）
• SyncToken（string、更新時に必須）：バージョン番号

明細項目：

{
  "Line": [
    {
      "Amount": 100.00,
      "DetailType": "SalesItemLineDetail",
      "SalesItemLineDetail": {
        "ItemRef": { "value": "1", "name": "Services" },
        "Qty": 1,
        "UnitPrice": 100.00,
        "TaxCodeRef": { "value": "TAX" }
      }
    },
    {
      "Amount": 100.00,
      "DetailType": "SubTotalLineDetail",
      "SubTotalLineDetail": {}
    }
  ]
}

支払い

請求書に対して顧客から受け取った支払いを表します。

主なフィールド：

• Id（string、読み取り専用）：一意の識別子
• TotalAmt（decimal、必須）：支払いの合計金額
• CustomerRef（object、必須）：顧客への参照
• PaymentMethodRef（object）：支払い方法（現金、小切手、クレジットカードなど）
• PaymentRefNum（string）：参照番号（小切手番号、取引 ID）
• TxnDate（date）：支払日
• DepositToAccountRef（object）：入金先の銀行口座
• Line（array）：請求書／クレジットメモへの支払いの適用
• UnappliedAmt（decimal、読み取り専用）：請求書に適用されていない金額
• SyncToken（string、更新時に必須）：バージョン番号

支払い明細項目（支払いを請求書に適用）：

{
  "Line": [
    {
      "Amount": 100.00,
      "LinkedTxn": [
        {
          "TxnId": "123",
          "TxnType": "Invoice"
        }
      ]
    }
  ]
}

Item

販売する商品またはサービスを表します。

タイプ:

• Service：サービス（コンサルティング、労務など）
• Inventory：在庫管理対象の実物商品
• NonInventory：在庫管理対象外の実物商品
• Category：他の項目をまとめるカテゴリ

主なフィールド:

• Id (string、読み取り専用)：一意の識別子
• Name (string、必須)：項目名（重複不可）
• Type (enum、必須)：Service、Inventory、NonInventory、Category
• Description (string)：項目の説明
• UnitPrice (decimal)：販売価格
• PurchaseCost (decimal)：仕入価格／原価
• IncomeAccountRef (object、必須)：収益勘定への参照
• ExpenseAccountRef (object)：仕入れに使用する費用勘定
• TrackQtyOnHand (boolean)：在庫数量を追跡するかどうか
• QtyOnHand (decimal)：現在の在庫数量
• Active (boolean)：項目が有効かどうか

Account

勘定科目表の勘定科目を表します。

主なフィールド:

• Id (string、読み取り専用)：一意の識別子
• Name (string、必須)：勘定科目名
• AccountType (enum、必須)：銀行、売掛金、買掛金、収益、費用など
• AccountSubType (enum)：より詳細な種類（CashOnHand、Checking、Savings など）
• CurrentBalance (decimal、読み取り専用)：現在の勘定残高
• Active (boolean)：勘定科目が有効かどうか
• Classification (enum)：資産、負債、資本、収益、費用

一般的な勘定科目の種類:

• Bank：銀行口座および現金勘定
• Accounts Receivable：顧客の残高
• Accounts Payable：仕入先の残高
• Income：収益勘定
• Expense：費用勘定
• Other Current Asset：短期資産
• Fixed Asset：長期資産

CRUD 操作のパターン

作成操作

最低限必要なフィールド：各エンティティには固有の必須フィールドがあります（通常は名前／参照と金額です）。

エンドポイントのパターン：POST /v3/company/{realmId}/{entityName}

リクエストヘッダー：

Authorization: Bearer {access_token}
Accept: application/json
Content-Type: application/json

Python の例 - 請求書の作成：

import requests

realm_id = "YOUR_REALM_ID"
access_token = "YOUR_ACCESS_TOKEN"

url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/invoice"

headers = {
    "Authorization": f"Bearer {access_token}",
    "Accept": "application/json",
    "Content-Type": "application/json"
}

invoice_data = {
    "Line": [
        {
            "Amount": 100.00,
            "DetailType": "SalesItemLineDetail",
            "SalesItemLineDetail": {
                "ItemRef": {"value": "1"}
            }
        }
    ],
    "CustomerRef": {"value": "1"}
}

response = requests.post(url, json=invoice_data, headers=headers)

if response.status_code == 200:
    invoice = response.json()['Invoice']
    print(f"Invoice created: {invoice['Id']}")
else:
    print(f"Error: {response.status_code} - {response.text}")

読み取り操作

単一エンティティ：GET /v3/company/{realmId}/{entityName}/{entityId}

Node.js の例 - 顧客の読み取り：

const axios = require('axios');

async function readCustomer(realmId, customerId, accessToken) {
  const url = `https://sandbox-quickbooks.api.intuit.com/v3/company/${realmId}/customer/${customerId}`;

  try {
    const response = await axios.get(url, {
      headers: {
        'Authorization': `Bearer ${accessToken}`,
        'Accept': 'application/json'
      }
    });

    return response.data.Customer;
  } catch (error) {
    if (error.response && error.response.status === 401) {
      // Token expired, refresh and retry
      console.error('Authentication failed - refresh token needed');
    } else {
      console.error('Read failed:', error.response?.data || error.message);
    }
    throw error;
  }
}

更新操作

更新には 2 種類あります。

1. 全件更新：書き込み可能なフィールドをすべて含める必要があります。省略したフィールドは NULL に設定されます。

2. スパース更新：指定したフィールドのみが更新されます。リクエスト本文に "sparse": true を設定します。

重要：常に最新の読み取りレスポンスに含まれる SyncToken を含めてください。これにより、同時変更の競合を防げます。

Python の例 - 顧客のメールアドレスのスパース更新：

import requests

def sparse_update_customer(realm_id, customer_id, sync_token, new_email, access_token):
    url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/customer"

    headers = {
        "Authorization": f"Bearer {access_token}",
        "Accept": "application/json",
        "Content-Type": "application/json"
    }

    # Sparse update - only updating email
    customer_data = {
        "Id": customer_id,
        "SyncToken": sync_token,
        "sparse": True,
        "PrimaryEmailAddr": {
            "Address": new_email
        }
    }

    response = requests.post(url, json=customer_data, headers=headers)

    if response.status_code == 200:
        updated_customer = response.json()['Customer']
        print(f"Customer updated, new SyncToken: {updated_customer['SyncToken']}")
        return updated_customer
    else:
        print(f"Update failed: {response.text}")
        return None

SyncToken の処理：

# 1. Read entity to get latest SyncToken
customer = read_customer(realm_id, customer_id, access_token)

# 2. Update with current SyncToken
updated = sparse_update_customer(
    realm_id,
    customer_id,
    customer['SyncToken'],  # Use current sync token
    "[email protected]",
    access_token
)

# 3. Store new SyncToken for next update
new_sync_token = updated['SyncToken']

削除操作

ほとんどのエンティティは、論理削除（Active を false に設定）または無効化操作を使用します。

論理削除のパターン：

// Mark customer as inactive
const deleteCustomer = {
  Id: customerId,
  SyncToken: currentSyncToken,
  sparse: true,
  Active: false
};

// POST to update endpoint
axios.post(`${baseUrl}/customer`, deleteCustomer, { headers });

物理削除（対象エンティティは限定的）： POST /v3/company/{realmId}/{entityName}?operation=delete

{
  "Id": "123",
  "SyncToken": "2"
}

クエリ言語とフィルタリング

QuickBooks は、制限付きの SQL 風クエリ構文を使用します。

クエリ構文

基本パターン：

SELECT * FROM {EntityName} WHERE {field} {operator} '{value}'

エンドポイント：GET /v3/company/{realmId}/query?query={sqlQuery}

演算子

• =：等しい
• <、>、<=、>=：比較
• IN：リスト内のいずれかの値に一致
• LIKE：パターンマッチング（ワイルドカード % のみサポートし、_ は使用できません）

例

名前で顧客を検索：

SELECT * FROM Customer WHERE DisplayName LIKE 'Acme%'

日付範囲で請求書を検索：

SELECT * FROM Invoice WHERE TxnDate >= '2024-01-01' AND TxnDate <= '2024-12-31'

並べ替えを指定してクエリ：

SELECT * FROM Customer WHERE Active = true ORDERBY DisplayName

ページネーション：

SELECT * FROM Invoice STARTPOSITION 1 MAXRESULTS 100

Python の例 - フィルターを使用したクエリ：

import requests
from urllib.parse import quote

def query_invoices_by_customer(realm_id, customer_id, access_token):
    query = f"SELECT * FROM Invoice WHERE CustomerRef = '{customer_id}' ORDERBY TxnDate DESC"
    encoded_query = quote(query)

    url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/query?query={encoded_query}"

    headers = {
        "Authorization": f"Bearer {access_token}",
        "Accept": "application/json"
    }

    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        result = response.json()['QueryResponse']
        invoices = result.get('Invoice', [])
        print(f"Found {len(invoices)} invoices")
        return invoices
    else:
        print(f"Query failed: {response.text}")
        return []

クエリの制限

• % 以外のワイルドカードは使用不可：LIKE でサポートされるのは % のみです（_ は使用できません）。
• JOIN 操作は不可：一度にクエリできるエンティティは 1 つです。
• 関数は限定的：集計関数（SUM、COUNT など）は使用できません。
• 最大 1000 件：それ以上の結果セットにはページネーションを使用してください。
• すべてのフィールドを返却：特定のフィールドを選択することはできません（常にすべてのフィールドが返されます）。

ページネーションのパターン

def query_all_customers(realm_id, access_token):
    all_customers = []
    start_position = 1
    max_results = 1000

    while True:
        query = f"SELECT * FROM Customer STARTPOSITION {start_position} MAXRESULTS {max_results}"
        encoded_query = quote(query)
        url = f"{base_url}/company/{realm_id}/query?query={encoded_query}"

        response = requests.get(url, headers={"Authorization": f"Bearer {access_token}"})
        result = response.json()['QueryResponse']

        customers = result.get('Customer', [])
        if not customers:
            break

        all_customers.extend(customers)

        # Check if more results exist
        if len(customers) < max_results:
            break

        start_position += max_results

    return all_customers

バッチ操作

バッチ操作を使うと、1 回の HTTP リクエストで複数の API 呼び出しを実行できます（最大 30 件）。

バッチリクエストの構造

エンドポイント：POST /v3/company/{realmId}/batch

リクエスト本文：

{
  "BatchItemRequest": [
    {
      "bId": "bid1",
      "operation": "create",
      "Customer": {
        "DisplayName": "New Customer 1"
      }
    },
    {
      "bId": "bid2",
      "operation": "update",
      "Invoice": {
        "Id": "123",
        "SyncToken": "1",
        "sparse": true,
        "EmailStatus": "NeedToSend"
      }
    },
    {
      "bId": "bid3",
      "operation": "query",
      "Query": "SELECT * FROM Customer WHERE Active = true MAXRESULTS 10"
    }
  ]
}

バッチ ID の追跡

各操作には、結果を追跡するための一意の bId（バッチ ID）があります。

レスポンス構造：

{
  "BatchItemResponse": [
    {
      "bId": "bid1",
      "Customer": {
        "Id": "456",
        "DisplayName": "New Customer 1"
      }
    },
    {
      "bId": "bid2",
      "Invoice": {
        "Id": "123",
        "SyncToken": "2"
      }
    },
    {
      "bId": "bid3",
      "QueryResponse": {
        "Customer": [...]
      }
    }
  ]
}

Node.js の例 - 顧客の一括更新

async function batchUpdateCustomers(realmId, customers, accessToken) {
  const batchItems = customers.map((customer, index) => ({
    bId: `customer_${index}`,
    operation: 'update',
    Customer: {
      Id: customer.Id,
      SyncToken: customer.SyncToken,
      sparse: true,
      Active: true  // Reactivate all customers
    }
  }));

  const url = `https://sandbox-quickbooks.api.intuit.com/v3/company/${realmId}/batch`;

  try {
    const response = await axios.post(url, {
      BatchItemRequest: batchItems
    }, {
      headers: {
        'Authorization': `Bearer ${accessToken}`,
        'Content-Type': 'application/json'
      }
    });

    const results = response.data.BatchItemResponse;

    // Process results by batch ID
    results.forEach(result => {
      if (result.Fault) {
        console.error(`Error for ${result.bId}:`, result.Fault);
      } else {
        console.log(`Success for ${result.bId}: Customer ${result.Customer.Id}`);
      }
    });

    return results;
  } catch (error) {
    console.error('Batch operation failed:', error.response?.data || error.message);
    throw error;
  }
}

バッチ操作のメリット

• API 呼び出しの削減：30 個の操作を 1 回のリクエストで実行でき、個別に 30 回リクエストする必要がありません
• レイテンシの低減：複数回の往復ではなく、1 回で完了します
• レート制限に優しい：レート制限では 1 回の API 呼び出しとしてカウントされます
• 操作ごとにアトミック：各操作は個別に成功または失敗します

バッチ操作の種類

• create：新しいエンティティを作成します
• update：既存のエンティティを更新します
• delete：エンティティを削除します
• query：クエリを実行します

エラー処理とトラブルシューティング

HTTP ステータスコード

• 200 OK：リクエストは成功しました（ただし、本文に <Fault> 要素が含まれる場合があります）
• 400 Bad Request：構文が無効、またはリクエストの形式が正しくありません
• 401 Unauthorized：アクセストークンが無効または期限切れです
• 403 Forbidden：権限が不足しているか、リソースへのアクセスが制限されています
• 404 Not Found：リソースが存在しません
• 429 Too Many Requests：レート制限を超過しました
• 500 Internal Server Error：サーバー側の問題です（1 回再試行してください）
• 503 Service Unavailable：サービスが一時的に利用できません（バックオフを適用して再試行してください）

フォールトの種類

200 OK の場合でも、レスポンスにフォールト要素が含まれることがあります。

{
  "Fault": {
    "Error": [
      {
        "Message": "Duplicate Name Exists Error",
        "Detail": "The name supplied already exists.",
        "code": "6240",
        "element": "Customer.DisplayName"
      }
    ],
    "type": "ValidationFault"
  },
  "time": "2024-12-09T10:30:00.000-08:00"
}

フォールトの種類：

1. ValidationFault：リクエストデータが無効、またはビジネスルールに違反しています

   • 対処法：リクエストのペイロードを修正し、必須フィールドを確認してください

2. SystemFault：サーバー側のエラーです

   • 対処法：リクエストを再試行してください。解決しない場合はサポートにお問い合わせください

3. AuthenticationFault：認証情報が無効です

   • 対処法：アクセストークンを更新し、再認証してください

4. AuthorizationFault：権限が不足しています

   • 対処法：OAuth スコープを確認し、ユーザーが管理者アクセス権を持っていることを確認してください

よくあるエラーコード

コード	エラー	解決策
6000	ビジネス検証エラー	TotalAmt と必須フィールドを確認してください
3200	古いオブジェクト（SyncToken の不一致）	エンティティを再読み込みし、最新の SyncToken を取得してください
3100	無効な参照	参照先のエンティティ（CustomerRef、ItemRef）が存在することを確認してください
6240	名前が重複しています	Customer/Item に一意の DisplayName を使用してください
610	オブジェクトが見つかりません	エンティティ ID が存在することを確認してください
4001	無効なトークン	アクセストークンを更新してください

SDK による例外処理

Java SDK の例外：

• ValidationException：検証フォールト
• ServiceException：サービスフォールト
• AuthenticationException：認証フォールト
• BadRequestException：ステータス 400
• InvalidTokenException：ステータス 401
• InternalServiceException：ステータス 500

Python の例外処理：

from intuitlib.exceptions import AuthClientError

try:
    response = requests.post(url, json=data, headers=headers)
    response.raise_for_status()

    # Check for fault in response body
    result = response.json()
    if 'Fault' in result:
        fault = result['Fault']
        print(f"Fault Type: {fault['type']}")
        for error in fault['Error']:
            print(f"  Code {error['code']}: {error['Message']}")
            print(f"  Element: {error.get('element', 'N/A')}")
        return None

    return result

except requests.exceptions.HTTPError as e:
    if e.response.status_code == 401:
        # Token expired, refresh
        print("Token expired, refreshing...")
        # Implement token refresh logic
    elif e.response.status_code == 429:
        # Rate limited, implement backoff
        print("Rate limited, backing off...")
    else:
        print(f"HTTP Error: {e.response.status_code}")
        print(f"Response: {e.response.text}")

except AuthClientError as e:
    print(f"Auth error: {str(e)}")

デバッグの方法

1. 200 の場合もレスポンス本文を確認する：成功したレスポンスにもフォールト要素が含まれることがあります
2. intuit_tid を記録する：迅速な解決のため、サポートへの問い合わせに含めてください
3. SyncToken を検証する：読み取り操作で取得した最新バージョンを常に使用してください
4. まずサンドボックスでテストする：開発にはサンドボックス企業を使用してください
5. 再試行ロジックを実装する：500/503 エラーには指数バックオフを適用してください
6. エラーの詳細を解析する：error.code、element、message フィールドを確認してください

指数バックオフを使用した再試行パターン

async function apiCallWithRetry(apiFunction, maxRetries = 3) {
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      return await apiFunction();
    } catch (error) {
      const status = error.response?.status;

      // Retry on server errors
      if (status >= 500 && status < 600 && attempt < maxRetries - 1) {
        const delay = Math.pow(2, attempt) * 1000; // 1s, 2s, 4s
        console.log(`Attempt ${attempt + 1} failed, retrying in ${delay}ms...`);
        await new Promise(resolve => setTimeout(resolve, delay));
        continue;
      }

      // Don't retry on client errors
      throw error;
    }
  }
}

変更の検出と Webhook

変更データキャプチャ（CDC）

CDC は、指定した期間内（最大 30 日間）に変更されたエンティティを返します。

エンドポイント：GET /v3/company/{realmId}/cdc?entities={entityList}&changedSince={dateTime}

パラメーター：

• entities：カンマ区切りのリスト（例："Invoice,Customer,Payment"）
• changedSince：ISO 8601 タイムスタンプ（例："2024-12-01T09:00:00-07:00"）

Python の例：

from datetime import datetime, timedelta
from urllib.parse import urlencode

def get_changed_entities(realm_id, entity_types, since_datetime, access_token):
    # Format: 2024-12-01T09:00:00-07:00
    changed_since = since_datetime.strftime('%Y-%m-%dT%H:%M:%S-07:00')

    params = {
        'entities': ','.join(entity_types),
        'changedSince': changed_since
    }

    url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/cdc"

    response = requests.get(
        url,
        params=params,
        headers={"Authorization": f"Bearer {access_token}"}
    )

    if response.status_code == 200:
        cdc_response = response.json()['CDCResponse']

        # Process changed entities
        for query_response in cdc_response:
            entity_type = query_response.get('QueryResponse', [{}])[0]

            for entity_name, entities in entity_type.items():
                if entities:
                    for entity in entities:
                        status = entity.get('status', 'Updated')
                        if status == 'Deleted':
                            print(f"Deleted {entity_name}: {entity['Id']}")
                        else:
                            print(f"Changed {entity_name}: {entity['Id']}")

        return cdc_response
    else:
        print(f"CDC request failed: {response.text}")
        return None

# Usage: Get all invoices and customers changed in last 24 hours
since = datetime.now() - timedelta(hours=24)
changes = get_changed_entities(realm_id, ['Invoice', 'Customer'], since, access_token)

レスポンス構造：

{
  "CDCResponse": [
    {
      "QueryResponse": [
        {
          "Invoice": [
            {
              "Id": "123",
              "MetaData": {
                "LastUpdatedTime": "2024-12-09T10:30:00-08:00"
              },
              "TotalAmt": 100.00,
              "Balance": 50.00
              // ... full invoice object
            }
          ]
        }
      ]
    },
    {
      "QueryResponse": [
        {
          "Customer": [
            {
              "Id": "456",
              "status": "Deleted"
            }
          ]
        }
      ]
    }
  ],
  "time": "2024-12-09T11:00:00.000-08:00"
}

CDC のベストプラクティス

• 短い期間でクエリする：レスポンスあたりの上限は 1000 エンティティです。1 時間ごと、または 1 日ごとに確認してください
• 最後の同期時刻を保存する：LastUpdatedTime を追跡し、changedSince パラメーターを設定してください
• 削除を処理する：status: "Deleted" のエンティティには ID のみが含まれます
• エンティティ全体を取得する：CDC は変更箇所だけでなく、ペイロード全体を返します
• Webhook と組み合わせる：リアルタイム処理には Webhook を使用し、CDC をバックアップとして利用してください

Webhook（リアルタイム通知）

Webhook は、データが変更されたときに HTTP POST 通知を送信します。

セットアップ：

1. 開発者ダッシュボードで Webhook URL を設定します
2. 通知を受信する POST エンドポイントを実装します
3. 1 秒以内に 200 OK を返します
4. 通知を非同期で処理します

通知ペイロード：

{
  "eventNotifications": [
    {
      "realmId": "123456789",
      "dataChangeEvent": {
        "entities": [
          {
            "name": "Invoice",
            "id": "145",
            "operation": "Create",
            "lastUpdated": "2024-12-09T10:30:00.000Z"
          },
          {
            "name": "Payment",
            "id": "456",
            "operation": "Update",
            "lastUpdated": "2024-12-09T10:31:00.000Z"
          },
          {
            "name": "Customer",
            "id": "789",
            "operation": "Merge",
            "lastUpdated": "2024-12-09T10:32:00.000Z",
            "deletedId": "788"
          }
        ]
      }
    }
  ]
}

Node.js Webhook ハンドラー：

const express = require('express');
const crypto = require('crypto');

const app = express();
app.use(express.json());

// Webhook endpoint
app.post('/webhooks/quickbooks', async (req, res) => {
  // Verify webhook signature (recommended)
  const signature = req.headers['intuit-signature'];
  const payload = JSON.stringify(req.body);

  // Return 200 immediately (process async)
  res.status(200).send('OK');

  // Process notifications asynchronously
  processWebhook(req.body).catch(console.error);
});

async function processWebhook(notification) {
  for (const event of notification.eventNotifications) {
    const realmId = event.realmId;

    for (const entity of event.dataChangeEvent.entities) {
      console.log(`${entity.operation} on ${entity.name} ID ${entity.id}`);

      // Fetch full entity data
      if (entity.operation !== 'Delete') {
        await fetchAndProcessEntity(realmId, entity.name, entity.id);
      } else {
        await handleEntityDeletion(realmId, entity.name, entity.id);
      }
    }
  }
}

async function fetchAndProcessEntity(realmId, entityType, entityId) {
  // Fetch full entity using read endpoint
  const url = `https://quickbooks.api.intuit.com/v3/company/${realmId}/${entityType.toLowerCase()}/${entityId}`;
  // ... implement fetch and processing logic
}

Webhook と CDC の選択マトリクス

ユースケース	推奨
リアルタイム同期	Webhook
定期同期（1 時間ごと／毎日）	CDC
初期データ読み込み	CDC
ダウンタイム後の再接続	CDC
大量の変更	CDC（通知のオーバーヘッドを削減）
低レイテンシの要件	Webhook
バックアップ／冗長化	両方（Webhook をプライマリ、CDC をバックアップとして使用）

組み合わせアプローチのパターン

class QuickBooksSync:
    def __init__(self):
        self.last_cdc_sync = self.load_last_sync_time()

    def handle_webhook(self, notification):
        """Process real-time webhook"""
        for entity in notification['dataChangeEvent']['entities']:
            self.process_entity_change(entity)

        # Update last known change time
        self.last_cdc_sync = datetime.now()
        self.save_last_sync_time()

    def periodic_cdc_sync(self):
        """Catch any missed changes"""
        changes = get_changed_entities(
            self.realm_id,
            ['Invoice', 'Customer', 'Payment'],
            self.last_cdc_sync,
            self.access_token
        )

        for entity in self.extract_entities(changes):
            if not self.entity_exists_locally(entity):
                # Missed by webhook, process now
                self.process_entity_change(entity)

        self.last_cdc_sync = datetime.now()
        self.save_last_sync_time()

ベストプラクティス

パフォーマンスの最適化

1. 一括変更にはバッチ操作を使用します

   • 1 つのリクエストに最大 30 件の操作をまとめます
   • API 呼び出しを減らし、スループットを向上させます
   • 例：顧客 30 件を個別に更新する代わりに、バッチで更新します

2. 同期には CDC または Webhook を実装します

   • すべてのエンティティを繰り返しポーリングするのは避けます
   • CDC は変更されたエンティティのみを返します
   • Webhook はポーリングなしでリアルタイム通知を提供します

3. スパース更新でペイロードを最小限に抑えます

   • 変更するフィールドのみを送信します
   • データ転送量と処理時間を削減します
   • 誤ってフィールドを上書きするのを防ぎます

4. 参照データをローカルにキャッシュします

   • 支払い方法、税コード、アカウントはめったに変更されません
   • 一度クエリし、TTL を設定してキャッシュします
   • 重複する API 呼び出しを削減します

5. 大きな結果セットはページネーションします

   • MAXRESULTS を使用してクエリ結果を制限します
   • メモリの問題を避けるため、バッチで処理します
   • 例：顧客を一度に 100 件クエリします

データの整合性

1. 更新には必ず SyncToken を使用します

   • 同時変更の競合を防ぎます
   • 更新前にエンティティを読み込み、最新のトークンを取得します
   • 3200 エラーが発生した場合は、再読み込みして再試行します

2. 同時変更を適切に処理します

   def safe_update(realm_id, customer_id, changes, access_token):
       max_attempts = 3
       for attempt in range(max_attempts):
           # Read latest version
           customer = read_customer(realm_id, customer_id, access_token)
   
           # Apply changes
           customer.update(changes)
           customer['sparse'] = True
   
           # Attempt update
           try:
               return update_customer(realm_id, customer, access_token)
           except SyncTokenError:
               if attempt == max_attempts - 1:
                   raise
               continue  # Retry with fresh SyncToken
   

3. API 呼び出し前に必須フィールドを検証します

   • まずローカルでビジネスルールを確認します
   • API による検証エラーを減らします
   • 例：請求書を作成する前に、顧客が存在することを確認します

4. 信頼性の高い追跡には Webhook と CDC を併用します

   • リアルタイム更新には Webhook を使用します
   • 見逃した変更に備え、定期的な CDC をバックアップとして使用します
   • 最終同期時刻を保存します

トークン管理

1. アクセストークンの有効期限は 3600 秒です

   • 有効期限が切れる前に自動更新を設定します
   • 安全のため、50 分経過時に更新します

2. リフレッシュトークンを事前に更新します

   class TokenManager {
     constructor() {
       this.refreshTimer = null;
     }
   
     scheduleRefresh(expiresIn) {
       // Refresh 5 minutes before expiration
       const refreshTime = (expiresIn - 300) * 1000;
   
       this.refreshTimer = setTimeout(() => {
         this.refreshAccessToken();
       }, refreshTime);
     }
   
     async refreshAccessToken() {
       try {
         const newTokens = await oauthClient.refresh();
         this.storeTokens(newTokens);
         this.scheduleRefresh(newTokens.expires_in);
       } catch (error) {
         // Refresh failed, need re-authentication
         this.handleReauthentication();
       }
     }
   }
   

3. 常に最新のリフレッシュトークンを使用します

   • 古いリフレッシュトークンは、新しいトークンの発行から 24 時間後に有効期限が切れます
   • 更新後、すぐにリフレッシュトークンを保存します
   • 古いリフレッシュトークンは決して使用しないでください

4. トークンを安全に保存します

   • データベースで暗号化します
   • バージョン管理にコミットしないでください
   • 開発環境では環境変数を使用します

5. 401 レスポンスを自動的に処理します

   def api_call_with_auto_refresh(api_function):
       try:
           return api_function()
       except Unauthorized401Error:
           # Attempt token refresh
           refresh_tokens()
           # Retry with new token
           return api_function()
   

API レート制限

1. 429 に対して指数バックオフを実装します

   def call_with_rate_limit_handling(api_function):
       max_retries = 5
       base_delay = 1
   
       for attempt in range(max_retries):
           try:
               return api_function()
           except RateLimitError as e:
               if attempt == max_retries - 1:
                   raise
   
               delay = base_delay * (2 ** attempt)  # 1s, 2s, 4s, 8s, 16s
               time.sleep(delay)
               continue
   

2. バッチ操作を使用して呼び出し回数を減らします

   • 1 回のバッチリクエストなら、個別の 30 回のリクエストと比べて 30 倍削減できます
   • バッチはレート制限上、1 回の API 呼び出しとしてカウントされます

3. レート制限ヘッダーを監視します（提供されている場合）

   • 一部のエンドポイントは、ヘッダーでレート制限情報を返します
   • 制限内に収まるよう使用状況を追跡します

複数通貨に関する考慮事項

1. 複数通貨が有効な場合は CurrencyRef が必須です

   {
     "Invoice": {
       "CurrencyRef": {
         "value": "USD",
         "name": "United States Dollar"
       }
     }
   }
   

2. 為替レートの処理

   • API が為替レートを自動的に適用します
   • ExchangeRate フィールドに適用された換算レートが表示されます
   • 自国通貨の金額は自動的に計算されます

3. ロケール固有の必須フィールド

   • フランス：カスタム取引番号が有効な場合、DocNumber が必須です
   • 英国：税の処理が異なります（VAT）
   • ロケール固有のドキュメントを確認してください

テストと開発

1. サンドボックス企業を使用する（開発者アカウントがあれば無料です）

   • developer.intuit.com で作成します
   • 本番データとは分離されています
   • API の全機能と同等の機能を利用できます

2. OAuth フローをエンドツーエンドでテストする

   • 認可 URL → コード交換 → トークン更新
   • トークンの有効期限切れへの対応をテストします
   • リフレッシュトークンのローテーションを確認します

3. Webhook エンドポイントを検証する

   • サンプルペイロードを使ってテストします
   • 1 秒未満の応答時間を確保します
   • Webhook の署名検証に対応します

4. 本番環境ですべての障害タイプに対応する

   • ValidationFault、SystemFault、AuthenticationFault、AuthorizationFault
   • エラーの詳細（コード、メッセージ、要素）をログに記録します
   • 適切な再試行ロジックを実装します

5. API 呼び出しとエラーを監視する

   • 成功率／失敗率を追跡します
   • エラー率の上昇時にアラートを発します
   • サポートへの問い合わせに備えて intuit_tid をログに記録します

一般的なワークフロー

ワークフロー 1：請求書の作成と送信

シナリオ：顧客の請求書を作成し、メールで送信します。

手順：

1. 顧客を照会または作成する

# Check if customer exists
customers = query_customers_by_email(realm_id, "[email protected]", access_token)

if not customers:
    # Create new customer
    customer = create_customer(realm_id, {
        "DisplayName": "Acme Corp",
        "PrimaryEmailAddr": {"Address": "[email protected]"},
        "BillAddr": {
            "Line1": "123 Main St",
            "City": "San Francisco",
            "CountrySubDivisionCode": "CA",
            "PostalCode": "94105"
        }
    }, access_token)
else:
    customer = customers[0]

customer_id = customer['Id']

2. 明細項目に使用する商品を照会する

# Get service item
query = "SELECT * FROM Item WHERE Type = 'Service' AND Name = 'Consulting'"
items = query_entity(realm_id, query, access_token)
service_item = items[0]

3. 明細項目を含む請求書を作成する

invoice_data = {
    "TxnDate": "2024-12-09",
    "DueDate": "2024-12-23",
    "CustomerRef": {"value": customer_id},
    "BillEmail": {"Address": "[email protected]"},
    "EmailStatus": "NeedToSend",  # Mark for email sending
    "Line": [
        {
            "Amount": 1500.00,
            "DetailType": "SalesItemLineDetail",
            "SalesItemLineDetail": {
                "ItemRef": {"value": service_item['Id']},
                "Qty": 10,
                "UnitPrice": 150.00,
                "TaxCodeRef": {"value": "NON"}  # Non-taxable
            },
            "Description": "Consulting services - December 2024"
        },
        {
            "Amount": 1500.00,
            "DetailType": "SubTotalLineDetail",
            "SubTotalLineDetail": {}
        }
    ]
}

invoice = create_invoice(realm_id, invoice_data, access_token)
print(f"Invoice {invoice['DocNumber']} created: ${invoice['TotalAmt']}")

4. 請求書メールを送信する（EmailStatus = "NeedToSend" の場合は自動送信）

# QuickBooks automatically sends email when EmailStatus is NeedToSend
# Alternatively, use send endpoint:
send_url = f"{base_url}/company/{realm_id}/invoice/{invoice['Id']}/send"
params = {"sendTo": "[email protected]"}

response = requests.post(send_url, params=params, headers=headers)
if response.status_code == 200:
    print(f"Invoice sent to {customer['PrimaryEmailAddr']['Address']}")

5. 応答とリンクされた取引を処理する

# Check invoice status
print(f"Invoice ID: {invoice['Id']}")
print(f"Balance: ${invoice['Balance']}")
print(f"Email Status: {invoice['EmailStatus']}")

# Track linked transactions
if 'LinkedTxn' in invoice:
    for linked in invoice['LinkedTxn']:
        print(f"Linked {linked['TxnType']}: {linked['TxnId']}")

ワークフロー 2：請求書に対する入金の記録

シナリオ：顧客が小切手で請求書を支払います。

手順：

1. DocNumber で請求書を照会する

def find_invoice_by_number(realm_id, doc_number, access_token):
    query = f"SELECT * FROM Invoice WHERE DocNumber = '{doc_number}'"
    invoices = query_entity(realm_id, query, access_token)

    if not invoices:
        raise ValueError(f"Invoice {doc_number} not found")

    return invoices[0]

invoice = find_invoice_by_number(realm_id, "1045", access_token)
customer_id = invoice['CustomerRef']['value']
balance = invoice['Balance']

2. 支払いエンティティを作成する

# Get payment method (Check)
payment_methods = query_entity(realm_id, "SELECT * FROM PaymentMethod WHERE Name = 'Check'", access_token)
payment_method_id = payment_methods[0]['Id']

payment_data = {
    "TotalAmt": balance,  # Pay full amount
    "CustomerRef": {"value": customer_id},
    "PaymentMethodRef": {"value": payment_method_id},
    "PaymentRefNum": "1234",  # Check number
    "TxnDate": "2024-12-09",
    "Line": [
        {
            "Amount": balance,
            "LinkedTxn": [
                {
                    "TxnId": invoice['Id'],
                    "TxnType": "Invoice"
                }
            ]
        }
    ]
}

payment = create_payment(realm_id, payment_data, access_token)

3. 残高の更新を確認する

# Re-read invoice to see updated balance
updated_invoice = read_invoice(realm_id, invoice['Id'], access_token)

print(f"Original balance: ${balance}")
print(f"Payment amount: ${payment['TotalAmt']}")
print(f"New balance: ${updated_invoice['Balance']}")
print(f"Unapplied payment amount: ${payment.get('UnappliedAmt', 0)}")

4. 一部入金に対応する

def apply_partial_payment(realm_id, invoice_id, payment_amount, customer_id, access_token):
    payment_data = {
        "TotalAmt": payment_amount,  # Less than invoice balance
        "CustomerRef": {"value": customer_id},
        "Line": [
            {
                "Amount": payment_amount,
                "LinkedTxn": [
                    {
                        "TxnId": invoice_id,
                        "TxnType": "Invoice"
                    }
                ]
            }
        ]
    }

    payment = create_payment(realm_id, payment_data, access_token)

    # Check unapplied amount
    if payment['UnappliedAmt'] > 0:
        print(f"Warning: ${payment['UnappliedAmt']} unapplied (overpayment or error)")

    return payment

5. 複数の請求書に入金を適用する

def pay_multiple_invoices(realm_id, invoice_ids, amounts, customer_id, total_paid, access_token):
    lines = []

    for invoice_id, amount in zip(invoice_ids, amounts):
        lines.append({
            "Amount": amount,
            "LinkedTxn": [{
                "TxnId": invoice_id,
                "TxnType": "Invoice"
            }]
        })

    payment_data = {
        "TotalAmt": total_paid,
        "CustomerRef": {"value": customer_id},
        "Line": lines
    }

    return create_payment(realm_id, payment_data, access_token)

# Example: Pay two invoices with single check
payment = pay_multiple_invoices(
    realm_id,
    ["145", "146"],  # Invoice IDs
    [100.00, 50.00],  # Amounts applied to each
    customer_id,
    150.00,  # Total check amount
    access_token
)

ワークフロー 3：顧客管理

シナリオ：顧客のライフサイクル全体を管理します。

1. 住所を含めて顧客を作成する

def create_customer_complete(realm_id, customer_info, access_token):
    customer_data = {
        "DisplayName": customer_info['display_name'],
        "GivenName": customer_info.get('first_name'),
        "FamilyName": customer_info.get('last_name'),
        "CompanyName": customer_info.get('company_name'),
        "PrimaryEmailAddr": {
            "Address": customer_info['email']
        },
        "PrimaryPhone": {
            "FreeFormNumber": customer_info.get('phone')
        },
        "BillAddr": {
            "Line1": customer_info['address_line1'],
            "City": customer_info['city'],
            "CountrySubDivisionCode": customer_info['state'],
            "PostalCode": customer_info['zip']
        },
        "ShipAddr": {
            "Line1": customer_info.get('ship_line1', customer_info['address_line1']),
            "City": customer_info.get('ship_city', customer_info['city']),
            "CountrySubDivisionCode": customer_info.get('ship_state', customer_info['state']),
            "PostalCode": customer_info.get('ship_zip', customer_info['zip'])
        }
    }

    return create_customer(realm_id, customer_data, access_token)

2. スパース更新でメールアドレスを変更する

def update_customer_email(realm_id, customer_id, new_email, access_token):
    # Read current customer
    customer = read_customer(realm_id, customer_id, access_token)

    # Sparse update - only email
    update_data = {
        "Id": customer_id,
        "SyncToken": customer['SyncToken'],
        "sparse": True,
        "PrimaryEmailAddr": {
            "Address": new_email
        }
    }

    return update_customer(realm_id, update_data, access_token)

3. 顧客の取引を照会する

def get_customer_transactions(realm_id, customer_id, access_token):
    transactions = {}

    # Query invoices
    invoice_query = f"SELECT * FROM Invoice WHERE CustomerRef = '{customer_id}'"
    transactions['invoices'] = query_entity(realm_id, invoice_query, access_token)

    # Query payments
    payment_query = f"SELECT * FROM Payment WHERE CustomerRef = '{customer_id}'"
    transactions['payments'] = query_entity(realm_id, payment_query, access_token)

    # Query estimates
    estimate_query = f"SELECT * FROM Estimate WHERE CustomerRef = '{customer_id}'"
    transactions['estimates'] = query_entity(realm_id, estimate_query, access_token)

    # Calculate totals
    total_invoiced = sum(inv['TotalAmt'] for inv in transactions['invoices'])
    total_paid = sum(pmt['TotalAmt'] for pmt in transactions['payments'])

    transactions['summary'] = {
        'total_invoiced': total_invoiced,
        'total_paid': total_paid,
        'balance': total_invoiced - total_paid
    }

    return transactions

4. AR アカウント参照を更新する

# Change default AR account for customer
def update_customer_ar_account(realm_id, customer_id, new_ar_account_id, access_token):
    customer = read_customer(realm_id, customer_id, access_token)

    update_data = {
        "Id": customer_id,
        "SyncToken": customer['SyncToken'],
        "sparse": True,
        "ARAccountRef": {
            "value": new_ar_account_id
        }
    }

    return update_customer(realm_id, update_data, access_token)

ワークフロー 4：バッチ同期処理

シナリオ：CDC とバッチ更新を使用して、変更されたエンティティを同期します。

1. CDC を使用して変更されたエンティティを取得する

from datetime import datetime, timedelta

def sync_changed_entities(realm_id, last_sync_time, access_token):
    # Get changes since last sync
    entity_types = ['Invoice', 'Customer', 'Payment', 'Item']
    changes = get_changed_entities(realm_id, entity_types, last_sync_time, access_token)

    return changes

2. 更新を含むバッチ処理を構築する

def build_batch_updates(changes):
    batch_items = []
    bid_counter = 0

    # Process each entity type
    for entity_type in ['Customer', 'Invoice', 'Payment']:
        entities = extract_entities_by_type(changes, entity_type)

        for entity in entities:
            if entity.get('status') == 'Deleted':
                # Skip deleted entities or handle separately
                continue

            # Example: Mark all invoices as reviewed
            if entity_type == 'Invoice':
                batch_items.append({
                    "bId": f"invoice_{bid_counter}",
                    "operation": "update",
                    "Invoice": {
                        "Id": entity['Id'],
                        "SyncToken": entity['SyncToken'],
                        "sparse": True,
                        "PrivateNote": f"Synced at {datetime.now().isoformat()}"
                    }
                })
                bid_counter += 1

    return batch_items

3. バッチを非同期で実行する

async def execute_batch_sync(realm_id, batch_items, access_token):
    # Split into batches of 30 (API limit)
    batch_size = 30
    results = []

    for i in range(0, len(batch_items), batch_size):
        batch_chunk = batch_items[i:i+batch_size]

        batch_request = {
            "BatchItemRequest": batch_chunk
        }

        url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/batch"

        response = await async_post(url, batch_request, {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json"
        })

        if response.status == 200:
            batch_response = response.json()
            results.extend(batch_response['BatchItemResponse'])
        else:
            print(f"Batch {i//batch_size + 1} failed: {response.text}")

    return results

4. バッチ ID ごとにバッチ応答を処理する

def process_batch_results(batch_results):
    success_count = 0
    error_count = 0
    errors = []

    for result in batch_results:
        bid = result['bId']

        if 'Fault' in result:
            error_count += 1
            fault = result['Fault']
            errors.append({
                'batch_id': bid,
                'error_code': fault['Error'][0]['code'],
                'message': fault['Error'][0]['Message']
            })
            print(f"Error in {bid}: {fault['Error'][0]['Message']}")
        else:
            success_count += 1
            # Extract updated entity
            entity_type = list(result.keys())[0]
            if entity_type != 'bId':
                entity = result[entity_type]
                print(f"Success {bid}: {entity_type} {entity['Id']} updated")

    summary = {
        'total': len(batch_results),
        'success': success_count,
        'errors': error_count,
        'error_details': errors
    }

    return summary

# Complete workflow
async def sync_workflow(realm_id, last_sync_time, access_token):
    # 1. Get changes via CDC
    changes = sync_changed_entities(realm_id, last_sync_time, access_token)

    # 2. Build batch updates
    batch_items = build_batch_updates(changes)

    if not batch_items:
        print("No changes to sync")
        return

    # 3. Execute batch
    results = await execute_batch_sync(realm_id, batch_items, access_token)

    # 4. Process results
    summary = process_batch_results(results)

    print(f"Sync complete: {summary['success']}/{summary['total']} successful")
    if summary['errors'] > 0:
        print(f"Errors encountered: {summary['errors']}")
        for error in summary['error_details']:
            print(f"  {error['batch_id']}: {error['message']}")

    return summary

コード例

例 1：OAuth2 トークンの更新（Node.js）

自動更新機能を備えたトークン管理の全体例：

const OAuthClient = require('intuit-oauth');

class QuickBooksAuth {
  constructor(clientId, clientSecret, redirectUri, environment = 'sandbox') {
    this.oauthClient = new OAuthClient({
      clientId: clientId,
      clientSecret: clientSecret,
      environment: environment,
      redirectUri: redirectUri
    });

    this.accessToken = null;
    this.refreshToken = null;
    this.tokenExpiry = null;
    this.refreshTimer = null;
  }

  // Store tokens after authorization
  async storeTokens(authResponse) {
    this.accessToken = authResponse.token.access_token;
    this.refreshToken = authResponse.token.refresh_token;

    // Calculate expiry time
    const expiresIn = authResponse.token.expires_in; // 3600 seconds
    this.tokenExpiry = Date.now() + (expiresIn * 1000);

    // Schedule automatic refresh (5 minutes before expiry)
    this.scheduleRefresh(expiresIn - 300);

    // Persist tokens to secure storage
    await this.saveToDatabase({
      access_token: this.accessToken,
      refresh_token: this.refreshToken,
      expiry: this.tokenExpiry
    });
  }

  // Schedule automatic token refresh
  scheduleRefresh(delaySeconds) {
    if (this.refreshTimer) {
      clearTimeout(this.refreshTimer);
    }

    this.refreshTimer = setTimeout(async () => {
      try {
        await this.refreshAccessToken();
      } catch (error) {
        console.error('Scheduled token refresh failed:', error);
        // Notify admin that re-authentication needed
        this.notifyReauthenticationNeeded();
      }
    }, delaySeconds * 1000);
  }

  // Refresh access token
  async refreshAccessToken() {
    try {
      // Set refresh token in client
      this.oauthClient.setToken({
        refresh_token: this.refreshToken
      });

      // Refresh
      const authResponse = await this.oauthClient.refresh();

      console.log('Token refreshed successfully');

      // Store new tokens
      await this.storeTokens(authResponse);

      return authResponse;

    } catch (error) {
      console.error('Token refresh failed:', error.originalMessage);

      // Check if refresh token is invalid
      if (error.error === 'invalid_grant') {
        console.error('Refresh token invalid - re-authentication required');
        this.accessToken = null;
        this.refreshToken = null;
        throw new Error('Re-authentication required');
      }

      throw error;
    }
  }

  // Get valid access token (refresh if needed)
  async getAccessToken() {
    // Check if token is about to expire (within 5 minutes)
    const bufferTime = 5 * 60 * 1000; // 5 minutes

    if (!this.accessToken || Date.now() >= (this.tokenExpiry - bufferTime)) {
      console.log('Token expired or expiring soon, refreshing...');
      await this.refreshAccessToken();
    }

    return this.accessToken;
  }

  // Make API call with automatic token refresh
  async apiCall(url, options = {}) {
    try {
      const token = await this.getAccessToken();

      const response = await fetch(url, {
        ...options,
        headers: {
          ...options.headers,
          'Authorization': `Bearer ${token}`,
          'Accept': 'application/json'
        }
      });

      // Handle 401 - token might have expired
      if (response.status === 401) {
        console.log('401 Unauthorized - refreshing token and retrying...');
        await this.refreshAccessToken();

        // Retry with new token
        const newToken = await this.getAccessToken();
        return fetch(url, {
          ...options,
          headers: {
            ...options.headers,
            'Authorization': `Bearer ${newToken}`,
            'Accept': 'application/json'
          }
        });
      }

      return response;

    } catch (error) {
      console.error('API call failed:', error);
      throw error;
    }
  }

  // Save tokens to database (implement based on your storage)
  async saveToDatabase(tokens) {
    // Example: Save to database
    // await db.tokens.update({ realmId }, tokens);
  }

  // Notify admin about re-auth requirement
  notifyReauthenticationNeeded() {
    // Example: Send email or notification
    console.error('Re-authentication required for QuickBooks integration');
  }
}

// Usage
const auth = new QuickBooksAuth(
  'YOUR_CLIENT_ID',
  'YOUR_CLIENT_SECRET',
  'https://yourapp.com/callback',
  'sandbox'
);

// After OAuth authorization
auth.storeTokens(authResponse);

// Make API calls - automatic token refresh
const response = await auth.apiCall(
  'https://sandbox-quickbooks.api.intuit.com/v3/company/123/customer/456',
  { method: 'GET' }
);

例 2：請求書の作成（Python）

明細項目と税を含む請求書作成の全体例：

import requests
from datetime import datetime, timedelta

class QuickBooksInvoice:
    def __init__(self, realm_id, access_token, base_url='https://sandbox-quickbooks.api.intuit.com'):
        self.realm_id = realm_id
        self.access_token = access_token
        self.base_url = base_url

    def create_invoice(self, customer_id, line_items, due_days=30, tax_code='TAX', memo=None):
        """
        Create an invoice with multiple line items

        Args:
            customer_id: QuickBooks customer ID
            line_items: List of dicts with 'item_id', 'quantity', 'unit_price', 'description'
            due_days: Days until due date
            tax_code: Tax code ('TAX' for taxable, 'NON' for non-taxable)
            memo: Customer memo

        Returns:
            Created invoice dict or None if error
        """
        # Calculate dates
        txn_date = datetime.now().strftime('%Y-%m-%d')
        due_date = (datetime.now() + timedelta(days=due_days)).strftime('%Y-%m-%d')

        # Build line items
        lines = []
        subtotal = 0

        for idx, item in enumerate(line_items, start=1):
            amount = item['quantity'] * item['unit_price']
            subtotal += amount

            lines.append({
                "LineNum": idx,
                "Amount": amount,
                "DetailType": "SalesItemLineDetail",
                "Description": item.get('description', ''),
                "SalesItemLineDetail": {
                    "ItemRef": {
                        "value": item['item_id']
                    },
                    "Qty": item['quantity'],
                    "UnitPrice": item['unit_price'],
                    "TaxCodeRef": {
                        "value": tax_code
                    }
                }
            })

        # Add subtotal line
        lines.append({
            "Amount": subtotal,
            "DetailType": "SubTotalLineDetail",
            "SubTotalLineDetail": {}
        })

        # Build invoice payload
        invoice_data = {
            "TxnDate": txn_date,
            "DueDate": due_date,
            "CustomerRef": {
                "value": customer_id
            },
            "Line": lines,
            "BillEmail": {},  # Will be populated from customer
            "EmailStatus": "NotSet"
        }

        # Add memo if provided
        if memo:
            invoice_data["CustomerMemo"] = {
                "value": memo
            }

        # Make API request
        url = f"{self.base_url}/v3/company/{self.realm_id}/invoice"

        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Accept": "application/json",
            "Content-Type": "application/json"
        }

        try:
            response = requests.post(url, json=invoice_data, headers=headers)
            response.raise_for_status()

            # Check for fault in response
            result = response.json()

            if 'Fault' in result:
                self._handle_fault(result['Fault'])
                return None

            invoice = result['Invoice']

            print(f"✓ Invoice {invoice['DocNumber']} created")
            print(f"  Customer: {invoice['CustomerRef']['value']}")
            print(f"  Total: ${invoice['TotalAmt']:.2f}")
            print(f"  Due: {invoice['DueDate']}")
            print(f"  Balance: ${invoice['Balance']:.2f}")

            return invoice

        except requests.exceptions.HTTPError as e:
            print(f"✗ HTTP Error: {e.response.status_code}")
            print(f"  Response: {e.response.text}")
            return None
        except Exception as e:
            print(f"✗ Error creating invoice: {str(e)}")
            return None

    def send_invoice(self, invoice_id, email_address):
        """Send invoice via email"""
        url = f"{self.base_url}/v3/company/{self.realm_id}/invoice/{invoice_id}/send"

        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Accept": "application/json"
        }

        params = {"sendTo": email_address}

        try:
            response = requests.post(url, params=params, headers=headers)
            response.raise_for_status()

            result = response.json()

            if 'Fault' in result:
                self._handle_fault(result['Fault'])
                return False

            invoice = result['Invoice']
            print(f"✓ Invoice {invoice['DocNumber']} sent to {email_address}")
            print(f"  Email Status: {invoice['EmailStatus']}")

            return True

        except Exception as e:
            print(f"✗ Error sending invoice: {str(e)}")
            return False

    def _handle_fault(self, fault):
        """Handle fault responses"""
        print(f"✗ Fault Type: {fault['type']}")
        for error in fault['Error']:
            print(f"  Error {error['code']}: {error['Message']}")
            if 'element' in error:
                print(f"  Element: {error['element']}")

# Usage example
invoice_manager = QuickBooksInvoice(realm_id='123456789', access_token='your_token')

# Create invoice with multiple items
invoice = invoice_manager.create_invoice(
    customer_id='42',
    line_items=[
        {
            'item_id': '1',
            'quantity': 10,
            'unit_price': 150.00,
            'description': 'Consulting services - December 2024'
        },
        {
            'item_id': '5',
            'quantity': 1,
            'unit_price': 500.00,
            'description': 'Project management - December 2024'
        }
    ],
    due_days=30,
    tax_code='TAX',  # or 'NON' for non-taxable
    memo='Thank you for your business!'
)

if invoice:
    # Send invoice via email
    invoice_manager.send_invoice(invoice['Id'], '[email protected]')

例 3：顧客のスパース更新（Node.js）

エラー処理を含むスパース更新パターンの例：

const axios = require('axios');

class QuickBooksCustomer {
  constructor(realmId, accessToken, baseUrl = 'https://sandbox-quickbooks.api.intuit.com') {
    this.realmId = realmId;
    this.accessToken = accessToken;
    this.baseUrl = baseUrl;
  }

  async readCustomer(customerId) {
    const url = `${this.baseUrl}/v3/company/${this.realmId}/customer/${customerId}`;

    try {
      const response = await axios.get(url, {
        headers: {
          'Authorization': `Bearer ${this.accessToken}`,
          'Accept': 'application/json'
        }
      });

      return response.data.Customer;
    } catch (error) {
      console.error('Read customer failed:', error.response?.data || error.message);
      throw error;
    }
  }

  async sparseUpdate(customerId, updates) {
    // First, read customer to get current SyncToken
    const customer = await this.readCustomer(customerId);

    // Build sparse update payload
    const updateData = {
      Id: customerId,
      SyncToken: customer.SyncToken,
      sparse: true,
      ...updates
    };

    const url = `${this.baseUrl}/v3/company/${this.realmId}/customer`;

    try {
      const response = await axios.post(url, updateData, {
        headers: {
          'Authorization': `Bearer ${this.accessToken}`,
          'Accept': 'application/json',
          'Content-Type': 'application/json'
        }
      });

      // Check for fault
      if (response.data.Fault) {
        this.handleFault(response.data.Fault);
        return null;
      }

      const updatedCustomer = response.data.Customer;

      console.log(`✓ Customer ${updatedCustomer.DisplayName} updated`);
      console.log(`  New SyncToken: ${updatedCustomer.SyncToken}`);

      return updatedCustomer;

    } catch (error) {
      if (error.response?.status === 400) {
        const fault = error.response.data.Fault;

        // Handle SyncToken mismatch
        if (fault.Error[0].code === '3200') {
          console.log('SyncToken mismatch - retrying with fresh token...');
          // Recursive retry with new token
          return this.sparseUpdate(customerId, updates);
        }
      }

      console.error('Update failed:', error.response?.data || error.message);
      throw error;
    }
  }

  // Example: Update email
  async updateEmail(customerId, newEmail) {
    return this.sparseUpdate(customerId, {
      PrimaryEmailAddr: {
        Address: newEmail
      }
    });
  }

  // Example: Update phone
  async updatePhone(customerId, newPhone) {
    return this.sparseUpdate(customerId, {
      PrimaryPhone: {
        FreeFormNumber: newPhone
      }
    });
  }

  // Example: Update billing address
  async updateBillingAddress(customerId, address) {
    return this.sparseUpdate(customerId, {
      BillAddr: {
        Line1: address.line1,
        City: address.city,
        CountrySubDivisionCode: address.state,
        PostalCode: address.zip
      }
    });
  }

  // Example: Deactivate customer
  async deactivateCustomer(customerId) {
    return this.sparseUpdate(customerId, {
      Active: false
    });
  }

  // Example: Update multiple fields at once
  async updateMultipleFields(customerId, updates) {
    const sparseUpdates = {};

    if (updates.email) {
      sparseUpdates.PrimaryEmailAddr = { Address: updates.email };
    }

    if (updates.phone) {
      sparseUpdates.PrimaryPhone = { FreeFormNumber: updates.phone };
    }

    if (updates.displayName) {
      sparseUpdates.DisplayName = updates.displayName;
    }

    if (updates.notes) {
      sparseUpdates.Notes = updates.notes;
    }

    return this.sparseUpdate(customerId, sparseUpdates);
  }

  handleFault(fault) {
    console.error(`✗ Fault Type: ${fault.type}`);
    fault.Error.forEach(error => {
      console.error(`  Error ${error.code}: ${error.Message}`);
      if (error.element) {
        console.error(`  Element: ${error.element}`);
      }
    });
  }
}

// Usage
const customerManager = new QuickBooksCustomer('123456789', 'your_access_token');

// Update email
await customerManager.updateEmail('42', '[email protected]');

// Update phone
await customerManager.updatePhone('42', '(555) 987-6543');

// Update address
await customerManager.updateBillingAddress('42', {
  line1: '456 New Street',
  city: 'San Francisco',
  state: 'CA',
  zip: '94105'
});

// Update multiple fields
await customerManager.updateMultipleFields('42', {
  email: '[email protected]',
  phone: '(555) 111-2222',
  displayName: 'Updated Customer Name',
  notes: 'VIP customer - priority support'
});

// Deactivate customer
await customerManager.deactivateCustomer('42');

例 4：フィルターを使用した照会（Python）

日付範囲と並べ替えを含む複雑な照会：

import requests
from urllib.parse import quote
from datetime import datetime, timedelta

class QuickBooksQuery:
    def __init__(self, realm_id, access_token, base_url='https://sandbox-quickbooks.api.intuit.com'):
        self.realm_id = realm_id
        self.access_token = access_token
        self.base_url = base_url

    def query(self, sql_query):
        """Execute SQL-like query"""
        encoded_query = quote(sql_query)
        url = f"{self.base_url}/v3/company/{self.realm_id}/query?query={encoded_query}"

        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Accept": "application/json"
        }

        try:
            response = requests.get(url, headers=headers)
            response.raise_for_status()

            result = response.json()

            if 'Fault' in result:
                print(f"Query error: {result['Fault']}")
                return []

            query_response = result.get('QueryResponse', {})

            # Extract entities (keys vary by entity type)
            for key, value in query_response.items():
                if key not in ['startPosition', 'maxResults', 'totalCount']:
                    return value if isinstance(value, list) else []

            return []

        except Exception as e:
            print(f"Query failed: {str(e)}")
            return []

    def query_invoices_by_date_range(self, start_date, end_date, customer_id=None):
        """Query invoices within date range, optionally filtered by customer"""
        query = f"SELECT * FROM Invoice WHERE TxnDate >= '{start_date}' AND TxnDate <= '{end_date}'"

        if customer_id:
            query += f" AND CustomerRef = '{customer_id}'"

        query += " ORDERBY TxnDate DESC"

        invoices = self.query(query)

        print(f"Found {len(invoices)} invoices between {start_date} and {end_date}")

        # Calculate totals
        total_amount = sum(inv['TotalAmt'] for inv in invoices)
        total_balance = sum(inv['Balance'] for inv in invoices)

        print(f"Total invoiced: ${total_amount:.2f}")
        print(f"Outstanding balance: ${total_balance:.2f}")

        return invoices

    def query_overdue_invoices(self, as_of_date=None):
        """Query invoices past due date"""
        if not as_of_date:
            as_of_date = datetime.now().strftime('%Y-%m-%d')

        query = f"SELECT * FROM Invoice WHERE Balance > '0' AND DueDate < '{as_of_date}' ORDERBY DueDate"

        invoices = self.query(query)

        print(f"Found {len(invoices)} overdue invoices as of {as_of_date}")

        # Group by customer
        by_customer = {}
        for inv in invoices:
            customer_id = inv['CustomerRef']['value']
            if customer_id not in by_customer:
                by_customer[customer_id] = {
                    'customer_name': inv['CustomerRef'].get('name', 'Unknown'),
                    'invoices': [],
                    'total_overdue': 0
                }

            by_customer[customer_id]['invoices'].append(inv)
            by_customer[customer_id]['total_overdue'] += inv['Balance']

        # Print summary
        for customer_id, data in by_customer.items():
            print(f"\nCustomer: {data['customer_name']}")
            print(f"  Overdue invoices: {len(data['invoices'])}")
            print(f"  Total overdue: ${data['total_overdue']:.2f}")

        return invoices

    def query_customers_by_balance(self, min_balance=0):
        """Query customers with balance greater than minimum"""
        query = f"SELECT * FROM Customer WHERE Balance > '{min_balance}' ORDERBY Balance DESC"

        customers = self.query(query)

        print(f"Found {len(customers)} customers with balance > ${min_balance}")

        total_ar = sum(cust['Balance'] for cust in customers)
        print(f"Total accounts receivable: ${total_ar:.2f}")

        return customers

    def query_items_by_type(self, item_type='Service'):
        """Query items by type (Service, Inventory, NonInventory, Category)"""
        query = f"SELECT * FROM Item WHERE Type = '{item_type}' AND Active = true ORDERBY Name"

        items = self.query(query)

        print(f"Found {len(items)} active {item_type} items")

        return items

    def search_customers_by_name(self, search_term):
        """Search customers by display name"""
        query = f"SELECT * FROM Customer WHERE DisplayName LIKE '%{search_term}%' ORDERBY DisplayName"

        customers = self.query(query)

        print(f"Found {len(customers)} customers matching '{search_term}'")

        for cust in customers:
            print(f"  {cust['DisplayName']} - Balance: ${cust['Balance']:.2f}")

        return customers

    def query_recent_payments(self, days=30):
        """Query payments from last N days"""
        start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')

        query = f"SELECT * FROM Payment WHERE TxnDate >= '{start_date}' ORDERBY TxnDate DESC"

        payments = self.query(query)

        print(f"Found {len(payments)} payments in last {days} days")

        total_received = sum(pmt['TotalAmt'] for pmt in payments)
        print(f"Total payments received: ${total_received:.2f}")

        return payments

# Usage
query_service = QuickBooksQuery(realm_id='123456789', access_token='your_token')

# Query invoices for date range
invoices = query_service.query_invoices_by_date_range(
    start_date='2024-01-01',
    end_date='2024-12-31'
)

# Query invoices for specific customer
customer_invoices = query_service.query_invoices_by_date_range(
    start_date='2024-01-01',
    end_date='2024-12-31',
    customer_id='42'
)

# Find overdue invoices
overdue = query_service.query_overdue_invoices()

# Find customers with high balances
high_balance_customers = query_service.query_customers_by_balance(min_balance=1000.00)

# Search for customer
customers = query_service.search_customers_by_name('Acme')

# Get recent payments
recent_payments = query_service.query_recent_payments(days=30)

例 5：バッチ処理（Node.js）

複数のエンティティの一括作成／更新：

const axios = require('axios');

class QuickBooksBatch {
  constructor(realmId, accessToken, baseUrl = 'https://sandbox-quickbooks.api.intuit.com') {
    this.realmId = realmId;
    this.accessToken = accessToken;
    this.baseUrl = baseUrl;
  }

  async executeBatch(batchItems) {
    const url = `${this.baseUrl}/v3/company/${this.realmId}/batch`;

    try {
      const response = await axios.post(url, {
        BatchItemRequest: batchItems
      }, {
        headers: {
          'Authorization': `Bearer ${this.accessToken}`,
          'Content-Type': 'application/json',
          'Accept': 'application/json'
        }
      });

      const results = response.data.BatchItemResponse;

      // Process results
      const summary = {
        total: results.length,
        success: 0,
        errors: 0,
        results: []
      };

      results.forEach(result => {
        if (result.Fault) {
          summary.errors++;
          console.error(`✗ Error for ${result.bId}:`);
          result.Fault.Error.forEach(err => {
            console.error(`  ${err.code}: ${err.Message}`);
          });
          summary.results.push({
            bId: result.bId,
            status: 'error',
            error: result.Fault
          });
        } else {
          summary.success++;
          // Extract entity from result
          const entityType = Object.keys(result).find(k => k !== 'bId');
          const entity = result[entityType];

          console.log(`✓ Success for ${result.bId}: ${entityType} ${entity.Id}`);
          summary.results.push({
            bId: result.bId,
            status: 'success',
            entityType: entityType,
            entity: entity
          });
        }
      });

      console.log(`\nBatch complete: ${summary.success}/${summary.total} successful`);

      return summary;

    } catch (error) {
      console.error('Batch operation failed:', error.response?.data || error.message);
      throw error;
    }
  }

  // Batch create customers
  async batchCreateCustomers(customers) {
    const batchItems = customers.map((customer, index) => ({
      bId: `customer_create_${index}`,
      operation: 'create',
      Customer: {
        DisplayName: customer.displayName,
        PrimaryEmailAddr: { Address: customer.email },
        PrimaryPhone: { FreeFormNumber: customer.phone },
        BillAddr: {
          Line1: customer.address,
          City: customer.city,
          CountrySubDivisionCode: customer.state,
          PostalCode: customer.zip
        }
      }
    }));

    return this.executeBatch(batchItems);
  }

  // Batch update invoices
  async batchUpdateInvoices(updates) {
    const batchItems = updates.map((update, index) => ({
      bId: `invoice_update_${index}`,
      operation: 'update',
      Invoice: {
        Id: update.id,
        SyncToken: update.syncToken,
        sparse: true,
        ...update.changes
      }
    }));

    return this.executeBatch(batchItems);
  }

  // Batch query multiple entities
  async batchQuery(queries) {
    const batchItems = queries.map((query, index) => ({
      bId: `query_${index}`,
      operation: 'query',
      Query: query.sql
    }));

    const results = await this.executeBatch(batchItems);

    // Extract query results
    const queryResults = {};
    results.results.forEach(result => {
      if (result.status === 'success' && result.entity.QueryResponse) {
        queryResults[result.bId] = result.entity.QueryResponse;
      }
    });

    return queryResults;
  }

  // Mixed batch operations
  async mixedBatch(operations) {
    const batchItems = operations.map((op, index) => {
      const item = {
        bId: `op_${index}_${op.type}`,
        operation: op.operation
      };

      // Add entity or query data
      if (op.operation === 'query') {
        item.Query = op.data;
      } else {
        item[op.entityType] = op.data;
      }

      return item;
    });

    return this.executeBatch(batchItems);
  }
}

// Usage Examples

const batchService = new QuickBooksBatch('123456789', 'your_access_token');

// Example 1: Batch create customers
const newCustomers = [
  {
    displayName: 'Acme Corp',
    email: '[email protected]',
    phone: '(555) 111-1111',
    address: '123 Main St',
    city: 'San Francisco',
    state: 'CA',
    zip: '94105'
  },
  {
    displayName: 'TechStart Inc',
    email: '[email protected]',
    phone: '(555) 222-2222',
    address: '456 Market St',
    city: 'San Francisco',
    state: 'CA',
    zip: '94103'
  }
];

const createResults = await batchService.batchCreateCustomers(newCustomers);

// Example 2: Batch update invoices (mark as sent)
const invoiceUpdates = [
  { id: '145', syncToken: '0', changes: { EmailStatus: 'NeedToSend' } },
  { id: '146', syncToken: '0', changes: { EmailStatus: 'NeedToSend' } },
  { id: '147', syncToken: '0', changes: { EmailStatus: 'NeedToSend' } }
];

const updateResults = await batchService.batchUpdateInvoices(invoiceUpdates);

// Example 3: Batch queries
const queries = [
  { sql: 'SELECT * FROM Customer WHERE Active = true MAXRESULTS 10' },
  { sql: 'SELECT * FROM Invoice WHERE Balance > 0 MAXRESULTS 10' },
  { sql: 'SELECT * FROM Payment MAXRESULTS 10' }
];

const queryResults = await batchService.batchQuery(queries);

// Example 4: Mixed batch operations
const mixedOps = [
  {
    type: 'create_customer',
    operation: 'create',
    entityType: 'Customer',
    data: {
      DisplayName: 'New Customer',
      PrimaryEmailAddr: { Address: '[email protected]' }
    }
  },
  {
    type: 'update_invoice',
    operation: 'update',
    entityType: 'Invoice',
    data: {
      Id: '145',
      SyncToken: '1',
      sparse: true,
      CustomerMemo: { value: 'Thank you!' }
    }
  },
  {
    type: 'query_items',
    operation: 'query',
    data: 'SELECT * FROM Item WHERE Type = \'Service\' MAXRESULTS 5'
  }
];

const mixedResults = await batchService.mixedBatch(mixedOps);

console.log(`Mixed batch: ${mixedResults.success}/${mixedResults.total} successful`);

例 6：入金の適用（Python）

複数の請求書に入金を適用します：

import requests

class QuickBooksPayment:
    def __init__(self, realm_id, access_token, base_url='https://sandbox-quickbooks.api.intuit.com'):
        self.realm_id = realm_id
        self.access_token = access_token
        self.base_url = base_url

    def create_payment(self, customer_id, total_amount, payment_method_id,
                       payment_ref_num, txn_date, invoice_applications):
        """
        Create payment and apply to one or more invoices

        Args:
            customer_id: QuickBooks customer ID
            total_amount: Total payment amount
            payment_method_id: Payment method ID
            payment_ref_num: Check number or transaction reference
            txn_date: Payment date (YYYY-MM-DD)
            invoice_applications: List of {'invoice_id': str, 'amount': float}

        Returns:
            Created payment dict or None if error
        """
        # Build line items for invoice applications
        lines = []
        total_applied = 0

        for application in invoice_applications:
            lines.append({
                "Amount": application['amount'],
                "LinkedTxn": [
                    {
                        "TxnId": application['invoice_id'],
                        "TxnType": "Invoice"
                    }
                ]
            })
            total_applied += application['amount']

        # Check for unapplied amount
        unapplied = total_amount - total_applied
        if unapplied < 0:
            print(f"Warning: Applied amount (${total_applied}) exceeds payment (${total_amount})")
            return None

        # Build payment payload
        payment_data = {
            "TotalAmt": total_amount,
            "CustomerRef": {
                "value": customer_id
            },
            "TxnDate": txn_date,
            "PaymentMethodRef": {
                "value": payment_method_id
            },
            "PaymentRefNum": payment_ref_num,
            "Line": lines
        }

        # Make API request
        url = f"{self.base_url}/v3/company/{self.realm_id}/payment"

        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Accept": "application/json",
            "Content-Type": "application/json"
        }

        try:
            response = requests.post(url, json=payment_data, headers=headers)
            response.raise_for_status()

            result = response.json()

            if 'Fault' in result:
                self._handle_fault(result['Fault'])
                return None

            payment = result['Payment']

            print(f"✓ Payment created: ID {payment['Id']}")
            print(f"  Customer: {payment['CustomerRef']['value']}")
            print(f"  Total Amount: ${payment['TotalAmt']:.2f}")
            print(f"  Applied Amount: ${total_applied:.2f}")
            print(f"  Unapplied Amount: ${payment.get('UnappliedAmt', 0):.2f}")
            print(f"  Reference: {payment.get('PaymentRefNum', 'N/A')}")

            # Show invoice applications
            for line in payment['Line']:
                if 'LinkedTxn' in line:
                    for linked in line['LinkedTxn']:
                        print(f"  Applied ${line['Amount']:.2f} to {linked['TxnType']} {linked['TxnId']}")

            return payment

        except requests.exceptions.HTTPError as e:
            print(f"✗ HTTP Error: {e.response.status_code}")
            print(f"  Response: {e.response.text}")
            return None
        except Exception as e:
            print(f"✗ Error creating payment: {str(e)}")
            return None

    def apply_payment_to_invoices(self, customer_id, check_number, check_amount,
                                   check_date, invoices):
        """
        Apply a single check payment to multiple invoices

        Args:
            customer_id: Customer ID
            check_number: Check number
            check_amount: Total check amount
            check_date: Check date
            invoices: List of {'id': str, 'amount_to_apply': float, 'balance': float}

        Returns:
            Payment dict or None
        """
        # Get check payment method ID
        payment_methods = self.query_payment_methods()
        check_method = next((pm for pm in payment_methods if pm['Name'].lower() == 'check'), None)

        if not check_method:
            print("Check payment method not found")
            return None

        # Build invoice applications
        applications = []
        total_to_apply = 0

        for invoice in invoices:
            amount = min(invoice['amount_to_apply'], invoice['balance'])
            applications.append({
                'invoice_id': invoice['id'],
                'amount': amount
            })
            total_to_apply += amount

            print(f"Will apply ${amount:.2f} to Invoice {invoice['id']}")

        # Check if payment covers all applications
        if total_to_apply > check_amount:
            print(f"Warning: Total applications (${total_to_apply}) exceeds check amount (${check_amount})")
            return None

        # Create payment
        payment = self.create_payment(
            customer_id=customer_id,
            total_amount=check_amount,
            payment_method_id=check_method['Id'],
            payment_ref_num=check_number,
            txn_date=check_date,
            invoice_applications=applications
        )

        return payment

    def apply_partial_payment(self, customer_id, payment_amount, payment_method_id,
                              invoice_id, partial_amount):
        """Apply partial payment to invoice"""
        if partial_amount > payment_amount:
            print("Partial amount cannot exceed total payment")
            return None

        applications = [{
            'invoice_id': invoice_id,
            'amount': partial_amount
        }]

        payment = self.create_payment(
            customer_id=customer_id,
            total_amount=payment_amount,
            payment_method_id=payment_method_id,
            payment_ref_num='',
            txn_date=datetime.now().strftime('%Y-%m-%d'),
            invoice_applications=applications
        )

        if payment and payment.get('UnappliedAmt', 0) > 0:
            print(f"\nNote: ${payment['UnappliedAmt']:.2f} remains unapplied")
            print("This amount can be applied to future invoices or refunded")

        return payment

    def query_payment_methods(self):
        """Get available payment methods"""
        from urllib.parse import quote

        query = "SELECT * FROM PaymentMethod"
        encoded_query = quote(query)

        url = f"{self.base_url}/v3/company/{self.realm_id}/query?query={encoded_query}"

        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Accept": "application/json"
        }

        response = requests.get(url, headers=headers)
        result = response.json()

        return result.get('QueryResponse', {}).get('PaymentMethod', [])

    def _handle_fault(self, fault):
        """Handle fault responses"""
        print(f"✗ Fault Type: {fault['type']}")
        for error in fault['Error']:
            print(f"  Error {error['code']}: {error['Message']}")
            if 'element' in error:
                print(f"  Element: {error['element']}")

# Usage Examples

payment_service = QuickBooksPayment(realm_id='123456789', access_token='your_token')

# Example 1: Apply single check to multiple invoices
payment = payment_service.apply_payment_to_invoices(
    customer_id='42',
    check_number='1234',
    check_amount=1500.00,
    check_date='2024-12-09',
    invoices=[
        {'id': '145', 'amount_to_apply': 1000.00, 'balance': 1000.00},
        {'id': '146', 'amount_to_apply': 500.00, 'balance': 750.00}
    ]
)

# Example 2: Partial payment on single invoice
partial_payment = payment_service.apply_partial_payment(
    customer_id='42',
    payment_amount=500.00,
    payment_method_id='1',  # Cash
    invoice_id='147',
    partial_amount=500.00  # Invoice balance is $1000, paying $500
)

# Example 3: Payment with unapplied amount (credit for future invoices)
credit_payment = payment_service.create_payment(
    customer_id='42',
    total_amount=2000.00,  # Customer pays $2000
    payment_method_id='1',
    payment_ref_num='',
    txn_date='2024-12-09',
    invoice_applications=[
        {'invoice_id': '145', 'amount': 1000.00}  # Only $1000 applied
    ]
    # $1000 remains unapplied as credit
)

API リファレンスのクイックリンク

Context7 ライブラリ：Context7 MCP を使って最新のドキュメントを取得します。

• ライブラリ ID：/websites/developer_intuit_app_developer_qbo
• 最新のコード例や API の変更を確認する際に使用します

公式リソース：

• QuickBooks API Explorer：https://developer.intuit.com/app/developer/qbo/docs/api/accounting/all-entities/account

  • サンドボックステストに対応したインタラクティブな API リファレンス
  • エンティティ別のドキュメントとリクエスト例

• Developer Dashboard：https://developer.intuit.com/app/developer/myapps

  • アプリ、キー、ウェブフックを管理します
  • サンドボックス会社を作成します

SDK：

• Node.js：https://github.com/intuit/intuit-oauth（OAuth）＋ API 呼び出し用の axios
• Python：https://github.com/intuit/intuit-oauth-python（OAuth）＋ requests
• Java：https://github.com/intuit/QuickBooks-V3-Java-SDK
• PHP：https://github.com/intuit/QuickBooks-V3-PHP-SDK
• C#/.NET：https://github.com/intuit/QuickBooks-V3-DotNET-SDK

一般的なエンドポイント：

• ベース URL（サンドボックス）：https://sandbox-quickbooks.api.intuit.com/v3/company/{realmId}
• ベース URL（本番環境）：https://quickbooks.api.intuit.com/v3/company/{realmId}
• トークンエンドポイント：https://oauth.platform.intuit.com/oauth2/v1/tokens/bearer
• OAuth 認証：https://appcenter.intuit.com/connect/oauth2

よくある問題のトラブルシューティング

SyncToken の不一致（エラー 3200）

症状：エンティティの更新時に「stale object error」が発生する

原因：リクエストの SyncToken が現在のバージョンと一致しません（同時変更）

解決方法：

def safe_update_with_retry(entity_id, updates, max_attempts=3):
    for attempt in range(max_attempts):
        try:
            # Read latest version
            entity = read_entity(entity_id)

            # Apply changes
            entity.update(updates)
            entity['sparse'] = True

            # Attempt update
            return update_entity(entity)

        except SyncTokenError as e:
            if attempt == max_attempts - 1:
                raise
            print(f"SyncToken mismatch, retrying... (attempt {attempt + 1})")
            continue

必須フィールドの不足（エラー 6000）

症状：「business validation error」または「required field missing」が発生する

原因：TotalAmt、CustomerRef などの必須フィールド、またはエンティティ固有の必須項目が不足しています

解決方法：

• エンティティ固有の必須フィールドについて API ドキュメントを確認します
• Payment では TotalAmt と CustomerRef が必須です
• Invoice では CustomerRef と Line 配列が必須です
• API 呼び出しの前に、データをローカルで検証します

よくある必須フィールド：

• Customer：DisplayName（一意である必要があります）
• Invoice：CustomerRef、Line（1 件以上）
• Payment：TotalAmt、CustomerRef
• Item：Name、Type、IncomeAccountRef（Service の場合）

OAuth トークンの有効期限切れ（401 Unauthorized）

症状：「invalid_token」または「token_expired」エラーが発生する

原因：アクセストークンの有効期限が切れています（3600 秒経過後）

解決方法：

async function apiCallWithAutoRefresh(apiFunction) {
  try {
    return await apiFunction();
  } catch (error) {
    if (error.response?.status === 401) {
      // Token expired, refresh
      await refreshAccessToken();
      // Retry with new token
      return await apiFunction();
    }
    throw error;
  }
}

予防策：

• トークンを定期的に更新します（50 分ごと）
• トークンの有効期限を保存し、リクエスト前に確認します
• 401 応答を自動的に処理します

無効な参照（エラー 3100）

症状：CustomerRef、ItemRef などを参照した際に「object not found」が発生する

原因：参照先のエンティティが存在しないか、削除されています

解決方法：

def validate_reference(entity_type, entity_id):
    """Verify entity exists before creating reference"""
    try:
        entity = read_entity(entity_type, entity_id)
        return True
    except NotFoundError:
        print(f"{entity_type} {entity_id} not found")
        return False

# Before creating invoice
if validate_reference('Customer', customer_id):
    if validate_reference('Item', item_id):
        create_invoice(customer_id, item_id)

レート制限（429 Too Many Requests）

症状：「throttle_limit_exceeded」またはステータスコード 429 が発生する

原因：API のレート制限を超過しています

解決方法 — ジッターを加えた指数バックオフ：

import time
import random

def api_call_with_backoff(api_function, max_retries=5):
    for attempt in range(max_retries):
        try:
            return api_function()
        except RateLimitError:
            if attempt == max_retries - 1:
                raise

            # Exponential backoff with jitter
            delay = (2 ** attempt) + random.uniform(0, 1)
            print(f"Rate limited, waiting {delay:.1f}s...")
            time.sleep(delay)

予防策：

• バッチ操作を使って呼び出し回数を減らします
• レート制限付きのリクエストキューを実装します
• 頻繁にアクセスする参照データをキャッシュします

バッチ操作の失敗

症状：バッチ内の一部の操作は失敗しますが、ほかの操作は成功する

原因：各バッチ操作は独立しているため、1 つの失敗がほかの操作に影響することはありません

解決方法：

function processBatchResults(results) {
  const failed = results.filter(r => r.Fault);
  const succeeded = results.filter(r => !r.Fault);

  console.log(`Batch: ${succeeded.length} success, ${failed.length} failed`);

  // Retry failed operations individually
  for (const failure of failed) {
    console.log(`Retrying ${failure.bId}...`);
    // Implement individual retry logic
  }

  return { succeeded, failed };
}

複数通貨の検証エラー

症状：「currency not enabled」または「exchange rate required」が発生する

原因：複数通貨機能が有効になっていないか、CurrencyRef が不足しています

解決方法：

{
  "Invoice": {
    "CurrencyRef": {
      "value": "USD",
      "name": "United States Dollar"
    },
    "ExchangeRate": 1.0
  }
}

確認事項：

• QuickBooks の会社設定で複数通貨が有効になっていることを確認します
• 複数通貨が有効な場合は、必ず CurrencyRef を含めます
• 外貨の場合、API が為替レートを自動的に計算します

Webhook が通知を受信しない

症状：Webhook エンドポイントを設定しているのに、POST リクエストを受信しない

原因：エンドポイントの問題、SSL の問題、または応答時間の遅さ

解決策：

1. エンドポイントにパブリックアクセスできることを確認します（localhost ではないこと）
2. HTTPS を使用します（Webhook では必須です）
3. 1 秒以内に応答します（すぐに 200 OK を返し、処理は非同期で行います）
4. サンプルペイロードでテストします：

   curl -X POST https://yourapp.com/webhooks/quickbooks \
     -H "Content-Type: application/json" \
     -d '{"eventNotifications":[]}'
   

5. 開発者ダッシュボードで Webhook のログを確認します

CDC レスポンス内の削除済みエンティティ

症状：status="Deleted" のエンティティに ID しか含まれていない

原因：CDC は削除済みエンティティについて最小限のデータを返します

解決策：

def process_cdc_changes(changes):
    for entity in changes:
        if entity.get('status') == 'Deleted':
            # Only ID available
            handle_deletion(entity['Id'])
        else:
            # Full entity data available
            process_entity_update(entity)

---

このスキルでは、QuickBooks Online API 統合に関する包括的なガイダンスを提供します。最新の API ドキュメントと変更については、Context7 でライブラリ ID /websites/developer_intuit_app_developer_qbo を使用してください。