Zendesk API インクリメンタルエクスポート：完全な開発者ガイド

外部システムをZendeskデータと同期させておくことは、終わりのないタスクのように感じられるかもしれません。データウェアハウスを構築している場合でも、分析を実行している場合でも、チケット情報をCRMに同期している場合でも、最後の更新以降に変更されたものだけをフェッチする信頼性の高い方法が必要です。Zendesk APIのインクリメンタルエクスポートは、まさにこの目的のために構築されています。

リアルタイムクエリ用に設計された標準のAPIエンドポイントとは異なり、インクリメンタルエクスポートは、バルクデータ同期専用に構築されています。特定の時点以降に作成または更新されたレコードをフェッチできるため、ETLパイプラインや定期的なデータ同期に最適です。

このガイドでは、ZendeskインクリメンタルエクスポートAPIを効果的に使用する方法を学びます。2つのページネーションアプローチ（カーソルベースと時間ベース）について説明し、完全なPython実装をウォークスルーし、本番データパイプラインのベストプラクティスを共有します。カスタムインフラストラクチャを構築せずにZendesk分析を取得する方法をお探しの場合は、eesel AIがネイティブのZendesk連携を提供し、データ同期を自動的に処理します。

ZendeskインクリメンタルエクスポートAPIとは？

インクリメンタルエクスポートAPIは、個々のレコードのルックアップではなく、バルクデータのエクスポート用に設計されたエンドポイントのセットです。標準のTickets APIは、特定のチケットのフェッチやフィルターを使用した検索に最適ですが、大規模なデータセットを効率的に同期するためには最適化されていません。

インクリメンタルエクスポートの仕組みは次のとおりです。開始時間（Unixエポックタイムスタンプとして）を指定すると、APIはその時間以降に作成または更新されたすべてのレコードを返します。次回の同期では、前の応答の終了時間またはカーソルを新しい開始点として使用します。これにより、変更されたデータのみをフェッチする効率的な同期ループが作成されます。

APIは、次の2つのページネーション方法をサポートしています。

• カーソルベースのページネーションは、不透明なカーソルポインタを使用して位置を追跡します。より一貫性があり、重複を排除し、可能な場合は推奨されるアプローチです。
• 時間ベースのページネーションは、開始タイムスタンプと終了タイムスタンプを使用します。すべてのインクリメンタルエンドポイントでサポートされていますが、複数のアイテムが同じタイムスタンプを共有している場合、重複したレコードが返される可能性があります。

チケット、ユーザー、組織、チケットイベント、Talk通話データ、Chat会話、ヘルプセンターの記事、およびカスタムオブジェクトレコードをインクリメンタルにエクスポートできます。包括的な分析を構築するチームのために、ZendeskチケットAPIに関するガイドでは、インクリメンタルエクスポートを補完する追加のエンドポイントについて説明します。

インクリメンタルエクスポートとSearch APIの使い分け

よくある質問は、チケットデータをフェッチするためにインクリメンタルエクスポートとSearch APIのどちらを使用するかということです。重要な違いは目的です。インクリメンタルエクスポートはデータ同期用に設計されており、Searchはクエリ用に設計されています。

次の場合にインクリメンタルエクスポートを使用します。

• データをデータウェアハウスまたは外部システムに同期している場合
• すべてのチケットの変更を確実に処理する必要がある場合
• 分析またはレポートパイプラインを構築している場合
• 変更のみをフェッチしてAPI呼び出しを最小限に抑えたい場合

次の場合にSearch APIを使用します。

• 複雑な条件に一致する特定のチケットを見つける必要がある場合
• エージェント用の検索インターフェイスを構築している場合
• 高度なフィルタリングでリアルタイムの結果が必要な場合

カーソルベースと時間ベースのページネーション：どちらを使用すべきですか？

Zendeskは両方のページネーション方法を提供していますが、カーソルベースは可能な場合は明らかに優れています。それらの違いを次に示します。

カーソルベースのページネーション（推奨）

カーソルベースのエクスポートは、不透明なポインタ（カーソル）を使用して、データセット内の位置を追跡します。start_timeを使用した最初の要求の後、後続の要求ではcursorパラメーターを使用します。

主な利点：

• アイテムがタイムスタンプを共有している場合でも、重複したレコードはありません
• より一貫性のある応答時間とペイロードサイズ
• 大規模なデータセットのパフォーマンスが向上
• より高いレート制限（チケットの場合は1分あたり10リクエスト、ユーザーの場合は20リクエスト、またはHigh Volume APIアドオンを使用すると60リクエスト）

サポートされているリソース：

• チケット
• ユーザー
• カスタムオブジェクトレコード

仕組み：

1. start_timeパラメーターを使用して最初の要求を行います
2. 応答からafter_cursorを抽出します
3. 次の要求にcursorパラメーターを使用します
4. end_of_streamがtrueになるまで繰り返します

時間ベースのページネーション

時間ベースのエクスポートは、Unixエポックタイムスタンプを使用してクエリウィンドウを定義します。各応答には、次の要求のstart_timeとして使用するend_timeが含まれています。

制限事項：

• 複数のアイテムが同じタイムスタンプを共有している場合、重複したレコードが返される可能性があります
• パフォーマンスの一貫性が低い
• より低いレート制限（1分あたり10リクエスト）

使用する場合：

• カーソルベースが利用できない場合（チケットイベント、組織、Talkデータ）
• 重複処理が重要ではない単純なスクリプトの場合

意思決定マトリックス

可能な限りカーソルベースを使用してください。時間ベースを使用する必要があるのは、リソースタイプでカーソルベースが利用できない場合のみです。

はじめに：認証と前提条件

コーディングを開始する前に、いくつかの設定が必要です。

必須：

• 管理者権限を持つZendeskアカウント
• APIトークン（管理センター > アプリと連携機能 > API > Zendesk APIで生成）
• Python 3.7以降がインストールされていること

Pythonパッケージ：

pip install requests python-dotenv

環境設定： .envファイルを作成して、資格情報を安全に保存します。

ZENDESK_SUBDOMAIN=your-subdomain
ZENDESK_EMAIL=your-email@company.com
ZENDESK_API_TOKEN=your-api-token

インクリメンタルエクスポートAPIは、Basic認証を使用します。メールアドレスと/tokenを組み合わせてユーザー名として渡し、APIトークンをパスワードとして渡します。

ステップバイステップ：カーソルベースのページネーションを使用したチケットのエクスポート

カーソルベースのページネーションを使用してチケットをエクスポートするための完全な実装をウォークスルーしましょう。このパターンは、ページネーション、レート制限、およびエラー回復を処理します。

コード

import os
import time
import requests
from requests.auth import HTTPBasicAuth
from dotenv import load_dotenv

load_dotenv()

class ZendeskIncrementalExport:
    def __init__(self):
        self.subdomain = os.getenv('ZENDESK_SUBDOMAIN')
        self.email = os.getenv('ZENDESK_EMAIL')
        self.api_token = os.getenv('ZENDESK_API_TOKEN')
        self.base_url = f"https://{self.subdomain}.zendesk.com/api/v2"
        self.auth = HTTPBasicAuth(f"{self.email}/token", self.api_token)

    def export_tickets(self, start_time=None, cursor=None):
        """
        カーソルベースのページネーションを使用してチケットをエクスポートします。

        Args:
            start_time: 最初のexportのUnixタイムスタンプ（最初の呼び出しに必要）
            cursor: 前の応答からのカーソル（後続の呼び出し用）
        """
        url = f"{self.base_url}/incremental/tickets/cursor.json"
        params = {}

        if cursor:
            params['cursor'] = cursor
        elif start_time:
            params['start_time'] = start_time
        else:
            raise ValueError("start_timeまたはcursorのいずれかを指定する必要があります")

        try:
            response = requests.get(url, auth=self.auth, params=params, timeout=30)

            if response.status_code == 429:
                # レート制限 - 指数バックオフを実装します
                retry_after = int(response.headers.get('Retry-After', 60))
                print(f"レート制限されています。{retry_after}秒待機しています...")
                time.sleep(retry_after)
                return self.export_tickets(start_time, cursor)

            response.raise_for_status()
            return response.json()

        except requests.exceptions.RequestException as e:
            print(f"リクエストに失敗しました：{e}")
            raise

    def full_export(self, start_time):
        """
        ページネーションを自動的に処理して、完全なエクスポートを実行します。
        """
        all_tickets = []
        cursor = None
        page_count = 0

        while True:
            data = self.export_tickets(
                start_time=start_time if cursor is None else None,
                cursor=cursor
            )

            tickets = data.get('tickets', [])
            all_tickets.extend(tickets)
            page_count += 1

            print(f"ページ{page_count}を取得しました：{len(tickets)}件のチケット")

            # 終わりに達したかどうかを確認します
            if data.get('end_of_stream'):
                print(f"エクスポートが完了しました。チケットの合計：{len(all_tickets)}")
                return {
                    'tickets': all_tickets,
                    'final_cursor': data.get('after_cursor'),
                    'pages': page_count
                }

            cursor = data.get('after_cursor')

            # レート制限を尊重します - リクエスト間にスリープします
            time.sleep(3)  # 20リクエスト/分 = 1リクエストあたり3秒

if __name__ == "__main__":
    exporter = ZendeskIncrementalExport()

    # 24時間前から開始します
    import datetime
    start_time = int((datetime.datetime.now() - datetime.timedelta(days=1)).timestamp())

    result = exporter.full_export(start_time)

    # 次の同期のために最後のカーソルを保存します
    print(f"次の実行のためにこのカーソルを保存します：{result['final_cursor']}")

主な実装の詳細

1分間の除外ウィンドウ： APIは、競合状態を防ぐために、ごく最近の1分間のデータを除外します。end_timeは、1分前よりも新しくなることはありません。それに応じて同期スケジュールを計画してください。

レート制限の処理： このコードは、429応答を受信したときに指数バックオフを実装します。Retry-Afterヘッダーは、待機する正確な時間を通知します。

カーソルの永続化： データの書き込みとアトミックに最後のカーソルを常に保存します。データを書き込んだ後、カーソルを保存する前にスクリプトがクラッシュした場合、次回の実行時に重複したレコードが処理されます。

削除されたチケットの除外： エクスポートに削除されたチケットを含めたくない場合は、exclude_deleted=trueをリクエストパラメーターに追加します。削除されたチケットは、完全に削除されてから90日間保持されるため、除外しない限りエクスポートに表示されます。

その他のインクリメンタルエクスポートエンドポイントの操作

エンドポイントURLと利用可能なページネーション方法が異なりますが、同じパターンが他のZendeskリソースにも適用されます。

ユーザーと組織

これらはチケットと同じパターンに従います。

url = f"{base_url}/incremental/users/cursor.json"

url = f"{base_url}/incremental/organizations.json"

チケットイベント

チケットイベントは時間ベースのみであり、チケットに加えられたすべての変更の記録が含まれています。

url = f"{base_url}/incremental/ticket_events.json"

params = {'start_time': start_time, 'include': 'comment_events'}

comment_eventsサイドロードは、変更に関するメタデータだけでなく、実際のコメントテキストが必要な場合に特に役立ちます。

Talk通話データ

音声分析のために、通話記録と通話レッグをエクスポートします。

url = f"{base_url}/channels/voice/stats/incremental/calls.json"

url = f"{base_url}/channels/voice/stats/incremental/legs.json"

レート制限：Talkエンドポイントの場合は1分あたり10リクエスト。

Chatデータ

Chatエクスポートでは、タイムスタンプに秒ではなくマイクロ秒を使用します。

url = f"{base_url}/chat/incremental/chats.json"

start_time_micro = start_time * 1000000
url = f"{base_url}/chat/incremental/agent_timeline.json?start_time={start_time_micro}"

注：Chat APIでは、開始時間は少なくとも5分前である必要があります。

ヘルプセンターの記事

記事のメタデータの変更をエクスポートします。

url = f"{base_url}/help_center/incremental/articles.json"

ページあたり最大1,000件の記事を返します。next_page URLには、最後の記事の更新タイムスタンプに基づいた新しい開始時間が含まれています。

レート制限とエラーの処理

本番データパイプラインには、堅牢なエラー処理が必要です。注意すべき点は次のとおりです。

レート制限ヘッダー

カーソルベースのエンドポイントは、詳細なレート制限情報を返します。

Zendesk-RateLimit-incremental-exports-cursor: total=20; remaining=15; resets=45

429応答を待つのではなく、これらのヘッダーを解析して、リクエストを事前に調整します。

指数バックオフ戦略

レート制限に達した場合は、ジッターを使用して指数バックオフを使用します。

import random

def backoff_with_jitter(attempt, base_delay=3):
    """指数バックオフとジッターを使用して遅延を計算します。"""
    delay = min(base_delay * (2 ** attempt), 60)  # 最大60秒
    jitter = random.uniform(0, delay * 0.1)  # 0〜10％のジッターを追加します
    return delay + jitter

for attempt in range(5):
    try:
        response = requests.get(url, auth=auth)
        if response.status_code == 429:
            delay = backoff_with_jitter(attempt)
            time.sleep(delay)
            continue
        response.raise_for_status()
        break
    except requests.exceptions.RequestException:
        if attempt == 4:  # 最後の試み
            raise
        delay = backoff_with_jitter(attempt)
        time.sleep(delay)

一般的なエラーと解決策

サンプルエクスポートを使用したテスト

Zendeskは、より厳しい制限（20分あたり10リクエスト）ですが、応答が小さいサンプルエクスポートエンドポイントを提供しています。これを開発およびテストに使用します。

url = f"{base_url}/incremental/tickets/sample.json?start_time={start_time}"

本番データパイプラインの構築

本番環境で使用するには、単純なスクリプトよりも堅牢なアーキテクチャが必要です。

推奨されるアーキテクチャ

スケジュールされたジョブ（Airflow/Lambda/Cron）
    ↓
インクリメンタルエクスポートAPI
    ↓
データの検証と変換
    ↓
データウェアハウス（Snowflake/BigQuery/Redshift）
    ↓
分析ダッシュボード

カーソルの状態の保存

本番環境でのカーソルの保存にローカルファイルを使用しないでください。永続的なストアを使用します。

import psycopg2

def save_cursor(cursor, last_sync_time):
    conn = psycopg2.connect(database_url)
    cur = conn.cursor()
    cur.execute("""
        INSERT INTO zendesk_sync_state (cursor, last_sync_time, updated_at)
        VALUES (%s, %s, NOW())
        ON CONFLICT (id) DO UPDATE SET
            cursor = EXCLUDED.cursor,
            last_sync_time = EXCLUDED.last_sync_time,
            updated_at = EXCLUDED.updated_at
    """, (cursor, last_sync_time))
    conn.commit()

時間ベースのエクスポートの重複排除

時間ベースのページネーションを使用している場合は、重複排除を実装します。

def deduplicate_records(records, key_fields):
    """
    複合キーに基づいて重複を削除します。
    チケットの場合：（id、updated_at）
    チケットイベントの場合：（id、created_at）
    """
    seen = set()
    unique = []

    for record in records:
        key = tuple(record.get(f) for f in key_fields)
        if key not in seen:
            seen.add(key)
            unique.append(record)

    return unique

tickets = deduplicate_records(tickets, ['id', 'updated_at'])

監視とアラート

パイプラインで次のメトリックを追跡します。

• 同期期間とレコード数
• レート制限ヒットと再試行回数
• 失敗したリクエストとエラー率
• データの鮮度（最後に成功した同期からの時間）

次のアラートを設定します。

• 同期の失敗または過度の再試行
• 異常に低いレコード数（APIの問題の可能性）
• SLAを超えるデータの鮮度

代替案：マネージドソリューション

データパイプラインの構築と維持には、エンジニアリングリソースが必要です。インフラストラクチャのオーバーヘッドなしでZendesk分析が必要な場合は、eesel AIのAIエージェントが、カスタムパイプラインを必要とせずに、Zendeskデータから直接、自動化されたチケット分析、レポート、およびインサイトを提供します。

主なベストプラクティスと制限事項

本番環境にデプロイする前に、次の点に注意してください。

ベストプラクティス

• 可能な場合は常にカーソルベースのページネーションを使用してください。 パフォーマンスと一貫性の利点はそれだけの価値があります。
• カーソルをデータの書き込みとアトミックに保存します。 トランザクションを使用して、カーソルとデータが常に同期していることを確認します。
• 1分間の除外ウィンドウを処理します。 1分前よりも新しいデータは期待しないでください。
• べき等書き込みを実装します。 宛先システムが重複したレコードを正常に処理できるように設計します。
• レート制限ヘッダーを監視します。 事前調整は、事後バックオフよりも優れています。

知っておくべき制限事項

• 削除されたチケットの保持： 削除されたチケットは、合計で約120日間エクスポートに残ります（完全に削除されるまで30日間、その後スクラブ後に90日間）。
• アーカイブされたチケット： Zendeskによってアーカイブされたチケットは、インクリメンタルエクスポートには含まれていません。
• スクラブされたデータ： 30日後、削除されたチケットの内容はスクラブされます（「SCRUBBED」または「X」に置き換えられます）。
• リアルタイムの保証はありません： APIは、リアルタイムストリーミングではなく、バッチ同期用に設計されています。

Zendeskデータの効率的な同期を開始する

Zendesk APIインクリメンタルエクスポートを使用すると、外部システムをサポートデータと同期させておくための信頼性の高い方法が得られます。カーソルベースのページネーションを使用し、レート制限を適切に処理し、状態を適切に保存することで、チケットボリュームに合わせて拡張できる堅牢なデータパイプラインを構築できます。

主なポイント：

• 可能な場合は、チケットとユーザーにカーソルベースのページネーションを使用します
• レート制限の処理に指数バックオフを実装します
• カーソルをデータの書き込みとアトミックに保存します
• 同期スケジュールで1分間のデータ除外ウィンドウを考慮します

特定のデータ変換のニーズがある場合、または独自のシステムと統合している場合は、カスタムパイプラインを構築することが理にかなっています。ただし、Zendeskデータからの分析、レポート、およびAI搭載のインサイトを主に探している場合は、eesel AIのようなマネージドソリューションがチームのエンジニアリング作業を数か月節約できるかどうかを検討してください。Zendesk連携はデータ同期を自動的に処理し、APIコードを1行も記述せずにチケット分析にすぐにアクセスできます。