Webhooks

Visão Geral

Os webhooks da DataSnap permitem que sua aplicação receba notificações automáticas sobre mudanças no status de processamento dos arquivos. Em vez de consultar constantemente a API para verificar o progresso, você recebe atualizações em tempo real diretamente em seus endpoints.

Benefícios dos webhooks: Reduz o número de requisições à API, fornece notificações instantâneas e permite arquiteturas orientadas a eventos.

Como Funciona

Configuração: Você configura URLs de webhook no painel da DataSnap
Processamento: Quando arquivos são processados, a DataSnap monitora mudanças de status
Notificação: A cada mudança de status, enviamos uma requisição POST para seus endpoints
Confirmação: Seu endpoint deve retornar status HTTP 2xx para confirmar o recebimento

Configuração de Webhooks

No Painel de Controle

Para configurar webhooks através da interface web:

Acessar a seção de Webhooks

No painel de controle da DataSnap, navegue até Webhooks no menu principal.

Criar novo webhook

Clique em ”+ Criar Webhook” para adicionar um novo endpoint.

Configurar endpoint

URL do Webhook: Informe a URL completa onde você quer receber as notificaçõesExemplo: https://sua-aplicacao.com/webhooks/datasnap

Salvar configuração

Clique em “Salvar” para ativar o webhook.

Teste seu webhook: Você pode enviar um webhook de teste através do painel para verificar se está funcionando corretamente.

Validação de URLs

Certifique-se de que seu endpoint webhook:

✅ Está acessível publicamente (não em localhost ou IPs privados)
✅ Aceita requisições POST
✅ Responde com status HTTP 2xx (200, 201, 204, etc.)
✅ Responde em até 10 segundos
✅ Usa HTTPS (recomendado para segurança)

Eventos de Webhook

Tipos de Eventos Suportados

Atualmente, o sistema envia webhooks para eventos relacionados ao processamento de arquivos:

Evento	Topic	Descrição
Aceito	`schemas.run.status`	Processamento foi aceito e adicionado à fila
Iniciado	`schemas.run.status`	Processamento começou a executar
Em Andamento	`schemas.run.status`	Processamento está sendo executado
Concluído	`schemas.run.status`	Processamento finalizado com sucesso
Falhou	`schemas.run.status`	Processamento falhou com erro

Payload do Webhook

Todas as notificações seguem o mesmo formato base:

{
  "topic": "schemas.run.status",
  "metadata": {
    "run_id": 24,
    "tenant": "sua_empresa",
    "status": "completed",
    "started_at": "2025-08-12T23:14:28-03:00"
  }
}

Campos do Payload

topic

string

Tópico do evento - Identifica o tipo de evento. Atualmente sempre schemas.run.status para processamento de arquivos.

metadata

object

Metadados do evento - Informações específicas sobre a execução do processamento.

Show Campos dos Metadados

run_id

integer

ID único da execução - Identificador exclusivo do processamento que pode ser usado para consultar detalhes via API.

tenant

string

Identificador do tenant - Nome da sua organização/conta na DataSnap.

status

string

Status atual - Estado do processamento: accepted, started, processing, completed, ou failed.

started_at

string

Timestamp de início - Data e hora em que o processamento foi iniciado, em formato ISO 8601.

completed_at

string

Timestamp de conclusão - Data e hora de conclusão (apenas em eventos completed ou failed).

error_message

string

Mensagem de erro - Descrição do erro que ocorreu (apenas em eventos failed).

files_processed

integer

Arquivos processados - Quantidade de arquivos que foram processados com sucesso (apenas em completed).

total_records

integer

Total de registros - Número total de registros processados (apenas em completed).

Exemplos de Payloads por Status

{
  "topic": "schemas.run.status",
  "metadata": {
    "run_id": 24,
    "tenant": "sua_empresa",
    "status": "accepted",
    "started_at": "2025-08-12T23:14:28-03:00"
  }
}

Implementação do Receptor

Estrutura Básica do Endpoint

Seu endpoint webhook deve:

Aceitar requisições POST
Processar o JSON do payload
Retornar status HTTP 2xx
Processar rapidamente (< 10 segundos)

from flask import Flask, request, jsonify
import json
import logging

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.route('/webhooks/datasnap', methods=['POST'])
def handle_datasnap_webhook():
    """Receptor de webhooks do DataSnap"""
    try:
        # Validar Content-Type
        if request.content_type != 'application/json':
            return jsonify({'error': 'Content-Type deve ser application/json'}), 400
        
        # Obter payload
        payload = request.get_json()
        
        if not payload:
            return jsonify({'error': 'Payload JSON inválido'}), 400
        
        # Log do evento recebido
        logger.info(f"Webhook recebido: {json.dumps(payload)}")
        
        # Processar baseado no tópico
        if payload.get('topic') == 'schemas.run.status':
            processar_status_execucao(payload['metadata'])
        else:
            logger.warning(f"Tópico desconhecido: {payload.get('topic')}")
        
        return jsonify({'status': 'processed'}), 200
        
    except Exception as e:
        logger.error(f"Erro ao processar webhook: {str(e)}")
        return jsonify({'error': 'Erro interno'}), 500

def processar_status_execucao(metadata):
    """Processa eventos de status de execução"""
    run_id = metadata.get('run_id')
    status = metadata.get('status')
    tenant = metadata.get('tenant')
    
    logger.info(f"Processamento {run_id} do tenant {tenant}: {status}")
    
    if status == 'completed':
        # Processamento concluído com sucesso
        files_processed = metadata.get('files_processed', 0)
        total_records = metadata.get('total_records', 0)
        
        logger.info(f"Sucesso: {files_processed} arquivos, {total_records} registros")
        
        # Aqui você pode:
        # - Atualizar banco de dados
        # - Enviar notificação por email
        # - Iniciar próximo processo na pipeline
        # - Atualizar dashboard em tempo real
        
    elif status == 'failed':
        # Processamento falhou
        error_message = metadata.get('error_message', 'Erro desconhecido')
        
        logger.error(f"Falha no processamento: {error_message}")
        
        # Aqui você pode:
        # - Registrar erro no sistema de monitoramento
        # - Enviar alerta para equipe técnica
        # - Agendar retry automático
        # - Notificar usuário sobre o problema
        
    elif status in ['accepted', 'started', 'processing']:
        # Processamento em andamento
        logger.info(f"Status intermediário: {status}")
        
        # Aqui você pode:
        # - Atualizar barra de progresso
        # - Mostrar status em tempo real no frontend
        # - Registrar tempo de início para métricas

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

Segurança

Validação de Origem

Para garantir que os webhooks realmente vêm da DataSnap:

Assinatura de Webhooks (Futuro)

Planejamos implementar assinatura HMAC para webhooks:

import hmac
import hashlib

def verificar_assinatura(payload_bytes, signature, secret):
    """Verifica assinatura HMAC do webhook (recurso futuro)"""
    expected = hmac.new(
        secret.encode('utf-8'),
        payload_bytes,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(f"sha256={expected}", signature)

Este recurso está em desenvolvimento. Por enquanto, use HTTPS para segurança.

Logs e Auditoria

Mantenha logs detalhados para auditoria:

def log_webhook_security(request, payload):
    """Log de segurança para webhooks"""
    security_info = {
        'timestamp': time.time(),
        'ip': request.remote_addr,
        'user_agent': request.headers.get('User-Agent'),
        'content_type': request.content_type,
        'payload_size': len(request.data),
        'run_id': payload.get('metadata', {}).get('run_id'),
        'tenant': payload.get('metadata', {}).get('tenant')
    }
    
    logger.info(f"WEBHOOK_SECURITY: {json.dumps(security_info)}")

Tratamento de Erros e Retry

Sistema de Retry da DataSnap

A DataSnap implementa retry automático para webhooks que falham:

Tentativas: Até 5 tentativas por webhook
Backoff: Exponencial com jitter (1s, 2s, 4s, 8s, 16s)
Timeout: 10 segundos por tentativa
Condições de retry: Status HTTP 5xx, timeout, erro de conexão

Status de Falha Permanente

Webhooks são marcados como falha permanente quando:

❌ Status HTTP 4xx: Erro do cliente (não será repetido)
❌ Após 5 tentativas: Todas as tentativas falharam
❌ URL inválida: URL não resolve ou não é acessível

Implementação de Idempotência

Seu endpoint deve ser idempotente para lidar com entregas duplicadas:

# Cache para evitar processamento duplicado
processed_webhooks = set()

@app.route('/webhooks/datasnap', methods=['POST'])
def handle_webhook():
    payload = request.get_json()
    
    # Criar ID único para o webhook
    webhook_id = f"{payload['metadata']['run_id']}_{payload['metadata']['status']}"
    
    if webhook_id in processed_webhooks:
        logger.info(f"Webhook duplicado ignorado: {webhook_id}")
        return jsonify({'status': 'already_processed'}), 200
    
    # Processar webhook
    try:
        processar_status_execucao(payload['metadata'])
        processed_webhooks.add(webhook_id)
        return jsonify({'status': 'processed'}), 200
    except Exception as e:
        # Não adicionar ao cache em caso de erro
        # para permitir retry
        raise

Monitoramento e Debugging

Logs Estruturados

Implemente logging estruturado para facilitar debugging:

import json
import logging
from datetime import datetime

class WebhookLogger:
    def __init__(self, name):
        self.logger = logging.getLogger(name)
        handler = logging.StreamHandler()
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)
    
    def log_webhook_received(self, payload, request_info):
        """Log quando webhook é recebido"""
        log_data = {
            'event': 'webhook_received',
            'timestamp': datetime.utcnow().isoformat(),
            'run_id': payload.get('metadata', {}).get('run_id'),
            'status': payload.get('metadata', {}).get('status'),
            'tenant': payload.get('metadata', {}).get('tenant'),
            'ip': request_info.get('ip'),
            'user_agent': request_info.get('user_agent')
        }
        self.logger.info(json.dumps(log_data))
    
    def log_processing_result(self, run_id, status, success, error=None):
        """Log resultado do processamento"""
        log_data = {
            'event': 'webhook_processed',
            'timestamp': datetime.utcnow().isoformat(),
            'run_id': run_id,
            'status': status,
            'success': success
        }
        
        if error:
            log_data['error'] = str(error)
        
        level = logging.INFO if success else logging.ERROR
        self.logger.log(level, json.dumps(log_data))

# Uso
webhook_logger = WebhookLogger('datasnap_webhooks')

Métricas de Webhook

Monitore estas métricas importantes:

from collections import defaultdict, Counter
import time

class WebhookMetrics:
    def __init__(self):
        self.counters = defaultdict(int)
        self.response_times = []
        self.errors = Counter()
        self.last_webhook = {}
    
    def record_webhook(self, payload, processing_time, success=True, error=None):
        """Registra métricas de um webhook"""
        status = payload.get('metadata', {}).get('status', 'unknown')
        tenant = payload.get('metadata', {}).get('tenant', 'unknown')
        
        # Contadores
        self.counters[f'webhooks_total'] += 1
        self.counters[f'webhooks_status_{status}'] += 1
        self.counters[f'webhooks_tenant_{tenant}'] += 1
        
        if success:
            self.counters[f'webhooks_success'] += 1
        else:
            self.counters[f'webhooks_failed'] += 1
            if error:
                self.errors[str(error)] += 1
        
        # Tempo de resposta
        self.response_times.append(processing_time)
        
        # Último webhook por tenant
        self.last_webhook[tenant] = time.time()
    
    def get_stats(self):
        """Retorna estatísticas atuais"""
        if not self.response_times:
            return self.counters
        
        return {
            **dict(self.counters),
            'avg_response_time': sum(self.response_times) / len(self.response_times),
            'max_response_time': max(self.response_times),
            'min_response_time': min(self.response_times),
            'total_webhooks_processed': len(self.response_times),
            'common_errors': dict(self.errors.most_common(5))
        }

# Instância global
metrics = WebhookMetrics()

Casos de Uso Avançados

Integração com Sistema de Notificações

def enviar_notificacao_usuario(metadata):
    """Envia notificação para usuário baseado no status"""
    run_id = metadata.get('run_id')
    status = metadata.get('status')
    tenant = metadata.get('tenant')
    
    # Buscar informações do usuário/projeto
    user_info = get_user_info_by_tenant(tenant)
    
    if status == 'completed':
        files_processed = metadata.get('files_processed', 0)
        total_records = metadata.get('total_records', 0)
        
        message = f"""
        ✅ Processamento concluído com sucesso!
        
        Arquivos processados: {files_processed}
        Registros processados: {total_records:,}
        ID da execução: {run_id}
        
        Seus dados estão prontos para consulta.
        """
        
        # Enviar email, SMS, push notification, etc.
        send_email(user_info['email'], "Processamento Concluído", message)
        send_push_notification(user_info['device_token'], message)
        
    elif status == 'failed':
        error_message = metadata.get('error_message', 'Erro desconhecido')
        
        message = f"""
        ❌ Falha no processamento
        
        ID da execução: {run_id}
        Erro: {error_message}
        
        Por favor, verifique seus arquivos e tente novamente.
        """
        
        send_email(user_info['email'], "Erro no Processamento", message)

Integração com Pipeline de Dados

def processar_pipeline_dados(metadata):
    """Integra webhook com pipeline de dados"""
    if metadata.get('status') == 'completed':
        run_id = metadata.get('run_id')
        tenant = metadata.get('tenant')
        
        # Disparar próximo processo na pipeline
        next_process_config = {
            'source_run_id': run_id,
            'tenant': tenant,
            'process_type': 'data_transformation',
            'triggered_by': 'webhook'
        }
        
        # Exemplo: adicionar job em fila
        redis_client.lpush('data_pipeline_queue', json.dumps(next_process_config))
        
        # Ou disparar função lambda/cloud function
        trigger_lambda_function('data-transformer', next_process_config)
        
        logger.info(f"Pipeline iniciada para run_id: {run_id}")

Dashboard em Tempo Real

# Integração com WebSocket para dashboard
import socketio

sio = socketio.Client()

def atualizar_dashboard(metadata):
    """Atualiza dashboard em tempo real via WebSocket"""
    dashboard_data = {
        'type': 'processing_update',
        'run_id': metadata.get('run_id'),
        'status': metadata.get('status'),
        'tenant': metadata.get('tenant'),
        'timestamp': time.time()
    }
    
    if metadata.get('status') == 'completed':
        dashboard_data.update({
            'files_processed': metadata.get('files_processed', 0),
            'total_records': metadata.get('total_records', 0),
            'completed_at': metadata.get('completed_at')
        })
    
    # Emitir para todos os clientes do tenant
    sio.emit('processing_update', dashboard_data, room=f"tenant_{metadata.get('tenant')}")

Troubleshooting

Problemas Comuns

Webhook não está sendo recebido

Possíveis causas:

URL não é acessível publicamente
Firewall bloqueando requisições
Endpoint não responde em tempo hábil
SSL/certificado inválido

Soluções:

Testar URL manualmente com curl
Verificar logs do servidor web
Confirmar que porta está aberta
Validar certificado SSL

# Teste manual do webhook
curl -X POST https://sua-app.com/webhooks/datasnap \
  -H "Content-Type: application/json" \
  -d '{
    "topic": "schemas.run.status",
    "metadata": {
      "run_id": 999,
      "tenant": "teste",
      "status": "completed"
    }
  }'

Webhooks duplicados

Causa: Sistema de retry da DataSnap pode causar entregas duplicadasSolução: Implementar idempotência no receptor

# Cache distribuído para ambientes multi-instância
import redis

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def is_webhook_processed(webhook_id):
    """Verifica se webhook já foi processado"""
    return redis_client.exists(f"webhook_processed:{webhook_id}")

def mark_webhook_processed(webhook_id, ttl=3600):
    """Marca webhook como processado (TTL de 1 hora)"""
    redis_client.setex(f"webhook_processed:{webhook_id}", ttl, "1")

Performance lenta

Sintoma: Timeout em webhooks ou processamento lentoSoluções:

Usar processamento assíncrono
Otimizar operações de banco de dados
Implementar cache
Usar filas para processamento pesado

# Processamento assíncrono com Celery
from celery import Celery

celery = Celery('webhook_processor')

@app.route('/webhooks/datasnap', methods=['POST'])
def handle_webhook():
    payload = request.get_json()
    
    # Processar assincronamente
    process_webhook_async.delay(payload)
    
    return jsonify({'status': 'accepted'}), 200

@celery.task
def process_webhook_async(payload):
    """Processa webhook em background"""
    # Processamento pesado aqui
    pass

Melhores Práticas

✅ Faça

Responda rapidamente: Confirme recebimento em < 2 segundos, processe depois
Seja idempotente: Lide com webhooks duplicados graciosamente
Use HTTPS: Para segurança das comunicações
Implemente logging: Para debugging e auditoria
Valide payloads: Sempre valide estrutura dos dados recebidos
Use filas: Para processamento assíncrono de tarefas pesadas

❌ Evite

Processamento síncrono longo: Não faça operações demoradas no endpoint
Ignorar erros HTTP: Sempre retorne status codes apropriados
Logs insuficientes: Dificulta debugging de problemas
Endpoints públicos sem segurança: Use filtros de IP ou autenticação
Falha silenciosa: Sempre log errors e notifique sobre problemas

Suporte e Recursos Adicionais

Precisa de ajuda com webhooks? Nossa equipe está disponível:

Documentação da API

Consulte os endpoints relacionados para mais detalhes técnicos.

Suporte Técnico

Entre em contato para dúvidas sobre implementação de webhooks.

Guia de Desenvolvimento

Veja exemplos de integração e boas práticas de desenvolvimento.

Limites e Quotas

Entenda os limites aplicados aos webhooks e como otimizar seu uso.

URLs de webhook são sensíveis: Certifique-se de que suas URLs estão protegidas e acessíveis apenas quando necessário.

Começando

Personalização

Escrevendo conteúdo

Ferramentas de IA

Visão Geral

Como Funciona

Configuração de Webhooks

No Painel de Controle

Validação de URLs

Eventos de Webhook

Tipos de Eventos Suportados

Payload do Webhook

Campos do Payload

Exemplos de Payloads por Status

Implementação do Receptor

Estrutura Básica do Endpoint

Segurança

Validação de Origem

Tratamento de Erros e Retry

Sistema de Retry da DataSnap

Status de Falha Permanente

Implementação de Idempotência

Monitoramento e Debugging

Logs Estruturados

Métricas de Webhook

Casos de Uso Avançados

Integração com Sistema de Notificações

Integração com Pipeline de Dados

Dashboard em Tempo Real

Troubleshooting

Problemas Comuns

Melhores Práticas

✅ Faça

❌ Evite

Suporte e Recursos Adicionais

Documentação da API

Suporte Técnico

Guia de Desenvolvimento

Limites e Quotas

Começando

Personalização

Escrevendo conteúdo

Ferramentas de IA

​Visão Geral

​Como Funciona

​Configuração de Webhooks

​No Painel de Controle

​Validação de URLs

​Eventos de Webhook

​Tipos de Eventos Suportados

​Payload do Webhook

​Campos do Payload

​Exemplos de Payloads por Status

​Implementação do Receptor

​Estrutura Básica do Endpoint

​Segurança

​Validação de Origem

​Tratamento de Erros e Retry

​Sistema de Retry da DataSnap

​Status de Falha Permanente

​Implementação de Idempotência

​Monitoramento e Debugging

​Logs Estruturados

​Métricas de Webhook

​Casos de Uso Avançados

​Integração com Sistema de Notificações

​Integração com Pipeline de Dados

​Dashboard em Tempo Real

​Troubleshooting

​Problemas Comuns

​Melhores Práticas

​✅ Faça

​❌ Evite

​Suporte e Recursos Adicionais

Documentação da API

Suporte Técnico

Guia de Desenvolvimento

Limites e Quotas

Visão Geral

Como Funciona

Configuração de Webhooks

No Painel de Controle

Validação de URLs

Eventos de Webhook

Tipos de Eventos Suportados

Payload do Webhook

Campos do Payload

Exemplos de Payloads por Status

Implementação do Receptor

Estrutura Básica do Endpoint

Segurança

Validação de Origem

Tratamento de Erros e Retry

Sistema de Retry da DataSnap

Status de Falha Permanente

Implementação de Idempotência

Monitoramento e Debugging

Logs Estruturados

Métricas de Webhook

Casos de Uso Avançados

Integração com Sistema de Notificações

Integração com Pipeline de Dados

Dashboard em Tempo Real

Troubleshooting

Problemas Comuns

Melhores Práticas

✅ Faça

❌ Evite

Suporte e Recursos Adicionais