Exportação incremental da API Zendesk: Um guia completo para desenvolvedores

Manter seus sistemas externos sincronizados com os dados do Zendesk pode parecer uma tarefa interminável. Seja construindo um data warehouse, executando análises ou sincronizando informações de tickets com seu CRM, você precisa de uma maneira confiável de buscar apenas o que foi alterado desde a última atualização. A exportação incremental da API Zendesk foi criada exatamente para esse propósito.

Ao contrário dos endpoints de API padrão projetados para consultas em tempo real, as exportações incrementais são construídas especificamente para sincronização de dados em massa. Elas permitem que você busque registros que foram criados ou atualizados desde um ponto específico no tempo, tornando-os ideais para pipelines ETL e sincronizações de dados regulares.

Neste guia, você aprenderá como usar a API de exportação incremental do Zendesk de forma eficaz. Abordaremos as duas abordagens de paginação (baseada em cursor e baseada em tempo), percorreremos uma implementação completa em Python e compartilharemos as melhores práticas para pipelines de dados de produção. Se você está procurando uma maneira de obter análises do Zendesk sem construir uma infraestrutura personalizada, o eesel AI oferece integrações nativas do Zendesk que lidam com a sincronização de dados automaticamente.

O que é a API de exportação incremental do Zendesk?

A API de exportação incremental é um conjunto de endpoints projetados para exportação de dados em massa, em vez de buscas de registros individuais. Embora a API padrão de Tickets seja ótima para buscar um ticket específico ou pesquisar com filtros, ela não é otimizada para sincronizar grandes conjuntos de dados de forma eficiente.

Veja como as exportações incrementais funcionam: você fornece um horário de início (como um timestamp Unix epoch) e a API retorna todos os registros criados ou atualizados desde esse horário. Na próxima sincronização, você usa o horário de término ou o cursor da resposta anterior como seu novo ponto de partida. Isso cria um loop de sincronização eficiente, onde você busca apenas os dados alterados.

A API suporta dois métodos de paginação:

• Paginação baseada em cursor usa um ponteiro de cursor opaco para rastrear a posição. É mais consistente, elimina duplicatas e é a abordagem recomendada quando disponível.
• Paginação baseada em tempo usa timestamps de início e fim. É suportada por todos os endpoints incrementais, mas pode retornar registros duplicados quando vários itens compartilham o mesmo timestamp.

Você pode exportar incrementalmente tickets, usuários, organizações, eventos de ticket, dados de chamadas do Talk, conversas do Chat, artigos da Central de Ajuda e registros de objetos personalizados. Para equipes que constroem análises abrangentes, nosso guia sobre a API de tickets do Zendesk cobre endpoints adicionais que complementam as exportações incrementais.

Quando usar exportações incrementais vs API de pesquisa

Uma pergunta comum é se deve usar exportações incrementais ou a API de pesquisa para buscar dados de tickets. A principal diferença é o propósito: as exportações incrementais são projetadas para sincronização de dados, enquanto a pesquisa é projetada para consultas.

Use exportações incrementais quando:

• Você está sincronizando dados com um data warehouse ou sistema externo
• Você precisa processar todas as alterações de tickets de forma confiável
• Você está construindo pipelines de análise ou relatórios
• Você deseja minimizar as chamadas de API buscando apenas as alterações

Use a API de pesquisa quando:

• Você precisa encontrar tickets específicos que correspondam a critérios complexos
• Você está construindo uma interface de pesquisa para agentes
• Você precisa de resultados em tempo real com filtragem avançada

Paginação baseada em cursor vs baseada em tempo: Qual você deve usar?

O Zendesk oferece ambos os métodos de paginação, mas o baseado em cursor é o claro vencedor quando disponível. Veja como eles diferem.

Paginação baseada em cursor (recomendada)

As exportações baseadas em cursor usam um ponteiro opaco (o cursor) para rastrear sua posição no conjunto de dados. Após sua solicitação inicial com um start_time, as solicitações subsequentes usam o parâmetro cursor.

Principais benefícios:

• Sem registros duplicados, mesmo quando os itens compartilham timestamps
• Tempos de resposta e tamanhos de payload mais consistentes
• Melhor desempenho para grandes conjuntos de dados
• Limites de taxa mais altos (10 solicitações por minuto para tickets, 20 para usuários ou 60 com o add-on High Volume API)

Recursos suportados:

• Tickets
• Usuários
• Registros de objetos personalizados

Como funciona:

1. Faça a solicitação inicial com o parâmetro start_time
2. Extraia after_cursor da resposta
3. Use o parâmetro cursor para a próxima solicitação
4. Repita até que end_of_stream seja verdadeiro

Paginação baseada em tempo

As exportações baseadas em tempo usam timestamps Unix epoch para definir sua janela de consulta. Cada resposta inclui um end_time que você usa como o start_time para sua próxima solicitação.

Limitações:

• Pode retornar registros duplicados quando vários itens compartilham o mesmo timestamp
• Desempenho menos consistente
• Limites de taxa mais baixos (10 solicitações por minuto)

Quando usar:

• Quando o baseado em cursor não está disponível (eventos de ticket, organizações, dados do Talk)
• Para scripts simples onde o tratamento de duplicatas não é crítico

Matriz de decisão

Use o baseado em cursor sempre que possível. A única vez que você deve usar o baseado em tempo é quando o baseado em cursor não está disponível para o seu tipo de recurso.

Começando: Autenticação e pré-requisitos

Antes de começar a codificar, você precisará configurar algumas coisas.

Requerido:

• Uma conta Zendesk com privilégios de administrador
• Um token de API (gere um em Admin Center > Apps and integrations > APIs > Zendesk API)
• Python 3.7+ instalado

Pacotes Python:

pip install requests python-dotenv

Configuração do ambiente: Crie um arquivo .env para armazenar suas credenciais com segurança:

ZENDESK_SUBDOMAIN=seu-subdomínio
ZENDESK_EMAIL=seu-email@empresa.com
ZENDESK_API_TOKEN=seu-token-api

A API de exportação incremental usa autenticação básica. Você passará seu endereço de e-mail combinado com /token como o nome de usuário e seu token de API como a senha.

Passo a passo: Exportando tickets com paginação baseada em cursor

Vamos percorrer uma implementação completa para exportar tickets usando paginação baseada em cursor. Este padrão lida com paginação, limitação de taxa e recuperação de erros.

O código

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):
        """
        Exportar tickets usando paginação baseada em cursor.

        Args:
            start_time: Timestamp Unix para exportação inicial (necessário para a primeira chamada)
            cursor: Cursor da resposta anterior (para chamadas subsequentes)
        """
        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:
                # Taxa limitada - implementar backoff exponencial
                retry_after = int(response.headers.get('Retry-After', 60))
                print(f"Taxa limitada. Esperando {retry_after} segundos...")
                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"Falha na solicitação: {e}")
            raise

    def full_export(self, start_time):
        """
        Realizar uma exportação completa, lidando com a paginação automaticamente.
        """
        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"Página buscada {page_count}: {len(tickets)} tickets")

            # Verifique se chegamos ao fim
            if data.get('end_of_stream'):
                print(f"Exportação completa. Total de tickets: {len(all_tickets)}")
                return {
                    'tickets': all_tickets,
                    'final_cursor': data.get('after_cursor'),
                    'pages': page_count
                }

            cursor = data.get('after_cursor')

            # Respeite os limites de taxa - espere entre as solicitações
            time.sleep(3)  # 20 solicitações/min = 1 solicitação a cada 3 segundos

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

    # Comece de 24 horas atrás
    import datetime
    start_time = int((datetime.datetime.now() - datetime.timedelta(days=1)).timestamp())

    result = exporter.full_export(start_time)

    # Salve o cursor final para a próxima sincronização
    print(f"Salve este cursor para a próxima execução: {result['final_cursor']}")

Principais detalhes da implementação

A janela de exclusão de 1 minuto: A API exclui dados do minuto mais recente para evitar condições de corrida. Seu end_time nunca será mais recente do que um minuto atrás. Planeje sua programação de sincronização de acordo.

Tratamento do limite de taxa: O código implementa backoff exponencial ao receber uma resposta 429. O cabeçalho Retry-After informa exatamente quanto tempo esperar.

Persistência do cursor: Sempre salve o cursor final atomicamente com suas gravações de dados. Se seu script falhar após gravar dados, mas antes de salvar o cursor, você processará registros duplicados na próxima execução.

Excluindo tickets excluídos: Adicione exclude_deleted=true aos parâmetros da sua solicitação se você não quiser tickets excluídos na sua exportação. Os tickets excluídos são retidos por 90 dias após a exclusão permanente, então eles aparecerão nas exportações, a menos que sejam excluídos.

Trabalhando com outros endpoints de exportação incremental

Os mesmos padrões se aplicam a outros recursos do Zendesk, embora os URLs dos endpoints e os métodos de paginação disponíveis variem.

Usuários e organizações

Estes seguem o mesmo padrão dos tickets:

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

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

Eventos de ticket

Os eventos de ticket são baseados em tempo apenas e incluem um registro de cada alteração feita nos tickets:

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

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

O sideload comment_events é particularmente útil se você precisar do texto real do comentário, não apenas dos metadados sobre as alterações.

Dados de chamadas do Talk

Exporte registros de chamadas e trechos de chamadas para análise de voz:

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

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

Limite de taxa: 10 solicitações por minuto para endpoints do Talk.

Dados do Chat

As exportações do Chat usam microssegundos em vez de segundos para timestamps:

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

Observação: a API do Chat exige que o horário de início seja de pelo menos 5 minutos no passado.

Artigos da Central de Ajuda

Exporte alterações de metadados de artigos:

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

Retorna até 1.000 artigos por página. O URL next_page contém um novo horário de início com base no timestamp de atualização do último artigo.

Lidando com limites de taxa e erros

Os pipelines de dados de produção precisam de tratamento de erros robusto. Veja o que observar.

Cabeçalhos de limite de taxa

Os endpoints baseados em cursor retornam informações detalhadas sobre o limite de taxa:

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

Analise esses cabeçalhos para limitar proativamente suas solicitações, em vez de esperar por respostas 429.

Estratégia de backoff exponencial

Quando você atingir um limite de taxa, use backoff exponencial com jitter:

import random

def backoff_with_jitter(attempt, base_delay=3):
    """Calcular atraso com backoff exponencial e jitter."""
    delay = min(base_delay * (2 ** attempt), 60)  # Limite em 60 segundos
    jitter = random.uniform(0, delay * 0.1)  # Adicione 0-10% de 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:  # Última tentativa
            raise
        delay = backoff_with_jitter(attempt)
        time.sleep(delay)

Erros comuns e soluções

Testando com exportações de amostra

O Zendesk fornece um endpoint de exportação de amostra com limites mais rigorosos (10 solicitações por 20 minutos), mas respostas menores. Use isso para desenvolvimento e teste:

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

Construindo um pipeline de dados de produção

Para uso em produção, você vai querer uma arquitetura mais robusta do que um script simples.

Arquitetura recomendada

Tarefa Agendada (Airflow/Lambda/Cron)
    ↓
API de Exportação Incremental
    ↓
Validação e Transformação de Dados
    ↓
Data Warehouse (Snowflake/BigQuery/Redshift)
    ↓
Painel de Análise

Armazenando o estado do cursor

Nunca confie em arquivos locais para armazenamento do cursor em produção. Use um armazenamento persistente:

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

Desduplicação para exportações baseadas em tempo

Se você estiver usando paginação baseada em tempo, implemente a desduplicação:

def deduplicate_records(records, key_fields):
    """
    Remover duplicatas com base na chave composta.
    Para tickets: (id, updated_at)
    Para eventos de ticket: (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'])

Monitoramento e alertas

Rastreie essas métricas em seu pipeline:

• Duração da sincronização e contagens de registros
• Acessos ao limite de taxa e contagens de repetição
• Solicitações com falha e taxas de erro
• Atualidade dos dados (tempo desde a última sincronização bem-sucedida)

Configure alertas para:

• Falhas de sincronização ou repetições excessivas
• Contagens de registros incomumente baixas (possíveis problemas de API)
• Atualidade dos dados excedendo seu SLA

Alternativa: Soluções gerenciadas

Construir e manter um pipeline de dados exige recursos de engenharia. Se sua equipe precisa de análises do Zendesk sem a sobrecarga da infraestrutura, o Agente de IA do eesel AI fornece análise automatizada de tickets, relatórios e insights diretamente dos seus dados do Zendesk, sem necessidade de pipeline personalizado.

Principais práticas recomendadas e limitações

Antes de implantar na produção, tenha esses pontos em mente.

Melhores práticas

• Sempre use paginação baseada em cursor quando disponível. Os benefícios de desempenho e consistência valem a pena.
• Armazene os cursores atomicamente com as gravações de dados. Use uma transação para garantir que o cursor e os dados estejam sempre sincronizados.
• Lide com a janela de exclusão de 1 minuto. Não espere dados mais recentes do que um minuto atrás.
• Implemente gravações idempotentes. Projete seu sistema de destino para lidar com registros duplicados normalmente.
• Monitore os cabeçalhos de limite de taxa. A limitação proativa é melhor do que o backoff reativo.

Limitações a serem conhecidas

• Retenção de tickets excluídos: Os tickets excluídos permanecem nas exportações por aproximadamente 120 dias no total (30 dias até a exclusão permanente e, em seguida, 90 dias após a limpeza).
• Tickets arquivados: Os tickets arquivados pelo Zendesk não são incluídos nas exportações incrementais.
• Dados limpos: Após 30 dias, os tickets excluídos têm seu conteúdo limpo (substituído por "SCRUBBED" ou "X").
• Sem garantias em tempo real: A API foi projetada para sincronizações em lote, não para streaming em tempo real.

Comece a sincronizar seus dados do Zendesk de forma eficiente

A exportação incremental da API Zendesk oferece uma maneira confiável de manter os sistemas externos sincronizados com seus dados de suporte. Ao usar a paginação baseada em cursor, lidar com os limites de taxa normalmente e armazenar o estado corretamente, você pode construir pipelines de dados robustos que escalam com seu volume de tickets.

Principais conclusões:

• Use paginação baseada em cursor para tickets e usuários quando possível
• Implemente backoff exponencial para tratamento do limite de taxa
• Armazene os cursores atomicamente com suas gravações de dados
• Considere a janela de exclusão de dados de 1 minuto em sua programação de sincronização

Construir um pipeline personalizado faz sentido quando você tem necessidades específicas de transformação de dados ou está integrando com sistemas proprietários. Mas se você está procurando principalmente análises, relatórios e insights baseados em IA dos seus dados do Zendesk, considere se uma solução gerenciada como o eesel AI pode economizar meses de esforço de engenharia para sua equipe. Nossa integração com o Zendesk lida com a sincronização de dados automaticamente, dando a você acesso imediato à análise de tickets sem escrever uma única linha de código da API.