Export incrémentiel de l'API Zendesk : un guide complet pour les développeurs

Garder vos systèmes externes synchronisés avec les données Zendesk peut sembler une tâche sans fin. Que vous construisiez un entrepôt de données, que vous exécutiez des analyses ou que vous synchronisiez les informations des tickets avec votre CRM, vous avez besoin d'un moyen fiable de récupérer uniquement ce qui a changé depuis votre dernière mise à jour. L'export incrémentiel de l'API Zendesk est conçu exactement dans ce but.

Contrairement aux points de terminaison API standard conçus pour les requêtes en temps réel, les exports incrémentiels sont conçus spécifiquement pour la synchronisation des données en masse. Ils vous permettent de récupérer les enregistrements qui ont été créés ou mis à jour depuis un moment précis, ce qui les rend idéaux pour les pipelines ETL et les synchronisations de données régulières.

Dans ce guide, vous apprendrez à utiliser efficacement l'API d'export incrémentiel de Zendesk. Nous aborderons les deux approches de pagination (basée sur le curseur et basée sur le temps), nous passerons en revue une implémentation Python complète et nous partagerons les meilleures pratiques pour les pipelines de données de production. Si vous cherchez un moyen d'obtenir des analyses Zendesk sans construire d'infrastructure personnalisée, eesel AI offre des intégrations Zendesk natives qui gèrent automatiquement la synchronisation des données.

Qu'est-ce que l'API d'export incrémentiel de Zendesk ?

L'API d'export incrémentiel est un ensemble de points de terminaison conçus pour l'export de données en masse plutôt que pour la recherche d'enregistrements individuels. Bien que l'API Tickets standard soit idéale pour récupérer un ticket spécifique ou effectuer une recherche avec des filtres, elle n'est pas optimisée pour la synchronisation efficace de grands ensembles de données.

Voici comment fonctionnent les exports incrémentiels : vous fournissez une heure de début (sous forme d'horodatage Unix epoch), et l'API renvoie tous les enregistrements créés ou mis à jour depuis cette heure. Lors de votre prochaine synchronisation, vous utilisez l'heure de fin ou le curseur de la réponse précédente comme nouveau point de départ. Cela crée une boucle de synchronisation efficace où vous ne récupérez que les données modifiées.

L'API prend en charge deux méthodes de pagination :

• La pagination basée sur le curseur utilise un pointeur de curseur opaque pour suivre la position. Elle est plus cohérente, élimine les doublons et est l'approche recommandée lorsqu'elle est disponible.
• La pagination basée sur le temps utilise des horodatages de début et de fin. Elle est prise en charge par tous les points de terminaison incrémentiels, mais peut renvoyer des enregistrements en double lorsque plusieurs éléments partagent le même horodatage.

Vous pouvez exporter de manière incrémentielle les tickets, les utilisateurs, les organisations, les événements de ticket, les données d'appel Talk, les conversations Chat, les articles du Centre d'aide et les enregistrements d'objets personnalisés. Pour les équipes qui construisent des analyses complètes, notre guide sur l'API des tickets Zendesk couvre des points de terminaison supplémentaires qui complètent les exports incrémentiels.

Quand utiliser les exports incrémentiels plutôt que l'API de recherche

Une question courante est de savoir s'il faut utiliser les exports incrémentiels ou l'API de recherche pour récupérer les données des tickets. La principale différence est l'objectif : les exports incrémentiels sont conçus pour la synchronisation des données, tandis que la recherche est conçue pour l'interrogation.

Utilisez les exports incrémentiels lorsque :

• Vous synchronisez des données vers un entrepôt de données ou un système externe
• Vous devez traiter de manière fiable chaque modification de ticket
• Vous construisez des pipelines d'analyse ou de reporting
• Vous voulez minimiser les appels API en ne récupérant que les modifications

Utilisez l'API de recherche lorsque :

• Vous devez trouver des tickets spécifiques correspondant à des critères complexes
• Vous construisez une interface de recherche pour les agents
• Vous avez besoin de résultats en temps réel avec un filtrage avancé

Pagination basée sur le curseur ou sur le temps : laquelle utiliser ?

Zendesk offre les deux méthodes de pagination, mais la pagination basée sur le curseur est clairement la meilleure lorsqu'elle est disponible. Voici comment elles diffèrent.

Pagination basée sur le curseur (recommandée)

Les exports basés sur le curseur utilisent un pointeur opaque (le curseur) pour suivre votre position dans l'ensemble de données. Après votre requête initiale avec un start_time, les requêtes suivantes utilisent le paramètre cursor.

Principaux avantages :

• Pas d'enregistrements en double, même lorsque les éléments partagent des horodatages
• Temps de réponse et tailles de payload plus cohérents
• Meilleures performances pour les grands ensembles de données
• Limites de débit plus élevées (10 requêtes par minute pour les tickets, 20 pour les utilisateurs, ou 60 avec l'extension API à volume élevé)

Ressources prises en charge :

• Tickets
• Utilisateurs
• Enregistrements d'objets personnalisés

Comment ça marche :

1. Effectuez la requête initiale avec le paramètre start_time
2. Extrayez after_cursor de la réponse
3. Utilisez le paramètre cursor pour la requête suivante
4. Répétez jusqu'à ce que end_of_stream soit vrai

Pagination basée sur le temps

Les exports basés sur le temps utilisent des horodatages Unix epoch pour définir votre fenêtre de requête. Chaque réponse inclut un end_time que vous utilisez comme start_time pour votre prochaine requête.

Limites :

• Peut renvoyer des enregistrements en double lorsque plusieurs éléments partagent le même horodatage
• Performances moins cohérentes
• Limites de débit plus faibles (10 requêtes par minute)

Quand l'utiliser :

• Lorsque la pagination basée sur le curseur n'est pas disponible (événements de ticket, organisations, données Talk)
• Pour les scripts simples où la gestion des doublons n'est pas essentielle

Matrice de décision

Utilisez la pagination basée sur le curseur chaque fois que possible. Le seul moment où vous devriez utiliser la pagination basée sur le temps est lorsque la pagination basée sur le curseur n'est pas disponible pour votre type de ressource.

Démarrage : authentification et prérequis

Avant de commencer à coder, vous devrez configurer quelques éléments.

Requis :

• Un compte Zendesk avec des privilèges d'administrateur
• Un jeton API (générez-en un dans Centre d'administration > Applications et intégrations > API > API Zendesk)
• Python 3.7+ installé

Packages Python :

pip install requests python-dotenv

Configuration de l'environnement : Créez un fichier .env pour stocker vos informations d'identification en toute sécurité :

ZENDESK_SUBDOMAIN=votre-sous-domaine
ZENDESK_EMAIL=votre-email@entreprise.com
ZENDESK_API_TOKEN=votre-jeton-api

L'API d'export incrémentiel utilise l'authentification de base. Vous passerez votre adresse e-mail combinée avec /token comme nom d'utilisateur, et votre jeton API comme mot de passe.

Étape par étape : exporter des tickets avec la pagination basée sur le curseur

Passons en revue une implémentation complète pour exporter des tickets en utilisant la pagination basée sur le curseur. Ce modèle gère la pagination, la limitation de débit et la récupération d'erreur.

Le code

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):
        """
        Export tickets using cursor-based pagination.

        Args:
            start_time: Unix timestamp for initial export (required for first call)
            cursor: Cursor from previous response (for subsequent calls)
        """
        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("Either start_time or cursor must be provided")

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

            if response.status_code == 429:
                # Rate limited - implement exponential backoff
                retry_after = int(response.headers.get('Retry-After', 60))
                print(f"Rate limited. Waiting {retry_after} seconds...")
                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"Request failed: {e}")
            raise

    def full_export(self, start_time):
        """
        Perform a complete export, handling pagination automatically.
        """
        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"Fetched page {page_count}: {len(tickets)} tickets")

            # Check if we've reached the end
            if data.get('end_of_stream'):
                print(f"Export complete. Total tickets: {len(all_tickets)}")
                return {
                    'tickets': all_tickets,
                    'final_cursor': data.get('after_cursor'),
                    'pages': page_count
                }

            cursor = data.get('after_cursor')

            # Respect rate limits - sleep between requests
            time.sleep(3)  # 20 requests/min = 1 request per 3 seconds

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

    # Start from 24 hours ago
    import datetime
    start_time = int((datetime.datetime.now() - datetime.timedelta(days=1)).timestamp())

    result = exporter.full_export(start_time)

    # Save the final cursor for next sync
    print(f"Save this cursor for next run: {result['final_cursor']}")

Principaux détails de l'implémentation

La fenêtre d'exclusion d'une minute : L'API exclut les données de la minute la plus récente pour éviter les conditions de concurrence. Votre end_time ne sera jamais plus récent qu'il y a une minute. Planifiez votre calendrier de synchronisation en conséquence.

Gestion de la limite de débit : Le code implémente un repli exponentiel lors de la réception d'une réponse 429. L'en-tête Retry-After vous indique exactement combien de temps attendre.

Persistance du curseur : Enregistrez toujours le curseur final de manière atomique avec vos écritures de données. Si votre script plante après avoir écrit des données mais avant d'enregistrer le curseur, vous traiterez les enregistrements en double lors de la prochaine exécution.

Exclusion des tickets supprimés : Ajoutez exclude_deleted=true à vos paramètres de requête si vous ne voulez pas de tickets supprimés dans votre export. Les tickets supprimés sont conservés pendant 90 jours après la suppression permanente, ils apparaîtront donc dans les exports à moins d'être exclus.

Travailler avec d'autres points de terminaison d'export incrémentiel

Les mêmes modèles s'appliquent aux autres ressources Zendesk, bien que les URL des points de terminaison et les méthodes de pagination disponibles varient.

Utilisateurs et organisations

Ceux-ci suivent le même modèle que les tickets :

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

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

Événements de ticket

Les événements de ticket sont basés uniquement sur le temps et incluent un enregistrement de chaque modification apportée aux tickets :

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

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

Le sideload comment_events est particulièrement utile si vous avez besoin du texte réel du commentaire, et pas seulement des métadonnées sur les modifications.

Données d'appel Talk

Exportez les enregistrements d'appels et les segments d'appels pour l'analyse vocale :

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

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

Limite de débit : 10 requêtes par minute pour les points de terminaison Talk.

Données de chat

Les exports de chat utilisent des microsecondes au lieu de secondes pour les horodatages :

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}"

Remarque : L'API Chat exige que l'heure de début soit au moins 5 minutes dans le passé.

Articles du Centre d'aide

Exportez les modifications des métadonnées des articles :

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

Renvoie jusqu'à 1 000 articles par page. L'URL next_page contient une nouvelle heure de début basée sur l'horodatage de la dernière mise à jour de l'article.

Gérer les limites de débit et les erreurs

Les pipelines de données de production ont besoin d'une gestion robuste des erreurs. Voici ce qu'il faut surveiller.

En-têtes de limite de débit

Les points de terminaison basés sur le curseur renvoient des informations détaillées sur la limite de débit :

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

Analysez ces en-têtes pour limiter proactivement vos requêtes plutôt que d'attendre les réponses 429.

Stratégie de repli exponentiel

Lorsque vous atteignez une limite de débit, utilisez un repli exponentiel avec jitter :

import random

def backoff_with_jitter(attempt, base_delay=3):
    """Calculate delay with exponential backoff and jitter."""
    delay = min(base_delay * (2 ** attempt), 60)  # Cap at 60 seconds
    jitter = random.uniform(0, delay * 0.1)  # Add 0-10% jitter
    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:  # Last attempt
            raise
        delay = backoff_with_jitter(attempt)
        time.sleep(delay)

Erreurs courantes et solutions

Tester avec des exports d'échantillon

Zendesk fournit un point de terminaison d'export d'échantillon avec des limites plus strictes (10 requêtes par 20 minutes) mais des réponses plus petites. Utilisez-le pour le développement et les tests :

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

Construire un pipeline de données de production

Pour une utilisation en production, vous aurez besoin d'une architecture plus robuste qu'un simple script.

Architecture recommandée

Tâche planifiée (Airflow/Lambda/Cron)
    ↓
API d'export incrémentiel
    ↓
Validation et transformation des données
    ↓
Entrepôt de données (Snowflake/BigQuery/Redshift)
    ↓
Tableau de bord d'analyse

Stocker l'état du curseur

Ne vous fiez jamais aux fichiers locaux pour le stockage du curseur en production. Utilisez un stockage persistant :

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()

Déduplication pour les exports basés sur le temps

Si vous utilisez la pagination basée sur le temps, implémentez la déduplication :

def deduplicate_records(records, key_fields):
    """
    Remove duplicates based on composite key.
    For tickets: (id, updated_at)
    For ticket events: (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'])

Surveillance et alertes

Suivez ces métriques dans votre pipeline :

• Durée de la synchronisation et nombre d'enregistrements
• Atteintes de la limite de débit et nombre de tentatives
• Requêtes ayant échoué et taux d'erreur
• Fraîcheur des données (temps écoulé depuis la dernière synchronisation réussie)

Configurez des alertes pour :

• Échecs de synchronisation ou tentatives excessives
• Nombre d'enregistrements inhabituellement bas (problèmes d'API possibles)
• Fraîcheur des données dépassant votre SLA (accord sur les niveaux de service)

Alternative : solutions gérées

La construction et la maintenance d'un pipeline de données nécessitent des ressources d'ingénierie. Si votre équipe a besoin d'analyses Zendesk sans les frais généraux d'infrastructure, l'agent d'IA d'eesel AI fournit une analyse automatisée des tickets, des rapports et des informations directement à partir de vos données Zendesk, sans pipeline personnalisé requis.

Principales pratiques exemplaires et limites

Avant de déployer en production, gardez ces points à l'esprit.

Pratiques exemplaires

• Utilisez toujours la pagination basée sur le curseur lorsque cela est possible. Les avantages en termes de performances et de cohérence en valent la peine.
• Stockez les curseurs de manière atomique avec les écritures de données. Utilisez une transaction pour vous assurer que le curseur et les données sont toujours synchronisés.
• Gérez la fenêtre d'exclusion d'une minute. Ne vous attendez pas à des données plus récentes qu'il y a une minute.
• Implémentez des écritures idempotentes. Concevez votre système de destination pour gérer les enregistrements en double avec élégance.
• Surveillez les en-têtes de limite de débit. La limitation proactive vaut mieux que le repli réactif.

Limites à connaître

• Rétention des tickets supprimés : Les tickets supprimés restent dans les exports pendant environ 120 jours au total (30 jours jusqu'à la suppression permanente, puis 90 jours après le nettoyage).
• Tickets archivés : Les tickets archivés par Zendesk ne sont pas inclus dans les exports incrémentiels.
• Données nettoyées : Après 30 jours, le contenu des tickets supprimés est nettoyé (remplacé par « SCRUBBED » ou « X »).
• Aucune garantie en temps réel : L'API est conçue pour les synchronisations par lots, pas pour le streaming en temps réel.

Commencez à synchroniser efficacement vos données Zendesk

L'export incrémentiel de l'API Zendesk vous offre un moyen fiable de maintenir les systèmes externes synchronisés avec vos données de support. En utilisant la pagination basée sur le curseur, en gérant les limites de débit avec élégance et en stockant correctement l'état, vous pouvez construire des pipelines de données robustes qui évoluent avec votre volume de tickets.

Principaux points à retenir :

• Utilisez la pagination basée sur le curseur pour les tickets et les utilisateurs lorsque cela est possible
• Implémentez un repli exponentiel pour la gestion des limites de débit
• Stockez les curseurs de manière atomique avec vos écritures de données
• Tenez compte de la fenêtre d'exclusion des données d'une minute dans votre calendrier de synchronisation

La construction d'un pipeline personnalisé est logique lorsque vous avez des besoins spécifiques de transformation des données ou que vous vous intégrez à des systèmes propriétaires. Mais si vous recherchez principalement des analyses, des rapports et des informations basées sur l'IA à partir de vos données Zendesk, déterminez si une solution gérée comme eesel AI pourrait faire économiser à votre équipe des mois d'efforts d'ingénierie. Notre intégration Zendesk gère automatiquement la synchronisation des données, vous donnant un accès immédiat à l'analyse des tickets sans écrire une seule ligne de code API.