Visão geral
Este guia apresenta as melhores práticas para integrar o DataSnap em suas aplicações, incluindo autenticação, tratamento de erros e padrões de integração recomendados.
Pré-requisitos :
Token de autenticação válido
Conhecimento básico de APIs RESTful
Familiaridade com formato JSONL
Autenticação
Token Bearer
Todas as requisições à API do DataSnap requerem autenticação via Bearer Token:
Authorization : Bearer SEU_TOKEN_AQUI
Obtendo seu token
Para obter seu token de acesso:
Acesse o painel de controle da DataSnap
Navegue até a seção “API” ou “Tokens de Acesso”
Clique em “Gerar Novo Token” ou “Criar Token”
Copie o token gerado e mantenha-o seguro
Cada token é associado a um tenant específico e possui permissões definidas.
Mantenha seu token seguro e nunca o compartilhe em repositórios públicos ou logs. Use variáveis de ambiente para armazená-lo.
Padrões de integração
Fluxo completo recomendado
Siga este padrão para uma integração robusta:
Validação de dados
Valide seus dados localmente antes do upload para evitar erros desnecessários. import json
def validar_jsonl ( arquivo ):
with open (arquivo, 'r' ) as f:
for linha_num, linha in enumerate (f, 1 ):
try :
obj = json.loads(linha.strip())
if 'data_evento' not in obj:
print ( f "Erro na linha { linha_num } : Campo 'data_evento' obrigatório ausente" )
return False
except json.JSONDecodeError as e:
print ( f "Erro na linha { linha_num } : { e } " )
return False
return True
Upload com retry
Implemente retry automático para uploads que podem falhar temporariamente. import time
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
def criar_sessao_com_retry ():
session = requests.Session()
retry = Retry(
total = 3 ,
status_forcelist = [ 500 , 502 , 503 , 504 ],
method_whitelist = [ "HEAD" , "GET" , "PUT" , "POST" , "DELETE" , "OPTIONS" , "TRACE" ],
backoff_factor = 1
)
adapter = HTTPAdapter( max_retries = retry)
session.mount( "http://" , adapter)
session.mount( "https://" , adapter)
return session
Tratamento de erros
Implemente tratamento robusto de erros para diferentes cenários:
def fazer_requisicao_api ( url , headers , data = None , files = None ):
try :
if files:
response = requests.post(url, headers = headers, files = files)
elif data:
response = requests.post(url, headers = headers, json = data)
else :
response = requests.get(url, headers = headers)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
status_code = e.response.status_code
if status_code == 401 :
print ( "Erro de autenticação. Verifique seu token." )
elif status_code == 404 :
print ( "Schema não encontrado." )
elif status_code == 422 :
print ( "Erro de validação:" , e.response.json())
else :
print ( f "Erro HTTP { status_code } : { e.response.text } " )
except requests.exceptions.ConnectionError:
print ( "Erro de conexão. Verifique sua internet." )
except requests.exceptions.Timeout:
print ( "Timeout na requisição. Tente novamente." )
except requests.exceptions.RequestException as e:
print ( f "Erro na requisição: { e } " )
Limites e quotas
A API DataSnap implementa limites para garantir performance estável e uso justo dos recursos para todos os usuários.
Resumo dos Principais Limites
Endpoint Limite Janela Upload de arquivos 10 MB por arquivo - Listagem de arquivos 100 requests 1 minuto Consultas 50 requests 1 minuto Geração de tokens 2 requests 1 minuto
A documentação de Limites e Quotas inclui exemplos de implementação de controle de requisições, tratamento de erros, monitoramento e muito mais.
Consultas otimizadas
Estrutura de consulta eficiente
Para melhor performance, siga estas práticas:
# ✅ Bom - consulta específica e limitada
consulta_otimizada = {
"select" : [ "id" , "nome" , "idade" ],
"where" : [
{ "field" : "idade" , "op" : ">=" , "value" : 18 },
{ "field" : "ativo" , "op" : "=" , "value" : True }
],
"order_by" : [{ "field" : "id" , "direction" : "asc" }],
"limit" : 100
}
# ❌ Evitar - consulta muito ampla
consulta_ineficiente = {
"select" : [ "*" ], # Evite select *
"limit" : 10000 # Limite muito alto
}
Paginação eficiente
Use cursor-based pagination para melhor performance:
def buscar_todos_registros ( schema_slug , headers , consulta_base ):
todos_registros = []
page_token = None
while True :
consulta = consulta_base.copy()
if page_token:
consulta[ "page_token" ] = page_token
response = requests.post(
f "https://api.datasnap.cloud/api/v1/schemas/ { schema_slug } /query" ,
headers = headers,
json = consulta
)
dados = response.json()
todos_registros.extend(dados[ "data" ])
# Verificar se há mais páginas
if "next_page_token" not in dados.get( "meta" , {}):
break
page_token = dados[ "meta" ][ "next_page_token" ]
return todos_registros
Monitoramento e logs
Implementando logs estruturados
import logging
import json
logging.basicConfig( level = logging. INFO )
logger = logging.getLogger( __name__ )
def log_api_call ( endpoint , method , status_code , response_time ):
log_data = {
"endpoint" : endpoint,
"method" : method,
"status_code" : status_code,
"response_time_ms" : response_time,
"timestamp" : time.time()
}
logger.info(json.dumps(log_data))
Métricas importantes
Monitore estas métricas em suas integrações:
Taxa de sucesso de uploads
Taxa de erro por endpoint
Latência de consultas
Tempo de resposta da API
Volume de dados consultados
Ambientes
URLs por ambiente
AMBIENTES = {
"producao" : "https://api.datasnap.cloud"
}
def obter_url_base ( ambiente = "producao" ):
return AMBIENTES .get(ambiente, AMBIENTES [ "producao" ])
Configuração por ambiente
Use variáveis de ambiente para configuração:
import os
class Config :
DATASNAP_TOKEN = os.getenv( "DATASNAP_TOKEN" )
DATASNAP_ENVIRONMENT = os.getenv( "DATASNAP_ENV" , "producao" )
DATASNAP_BASE_URL = obter_url_base( DATASNAP_ENVIRONMENT )
@ classmethod
def validate ( cls ):
if not cls . DATASNAP_TOKEN :
raise ValueError ( "DATASNAP_TOKEN não configurado" )
Próximos passos
Exemplos práticos Veja exemplos completos de integração na documentação da API.
Suporte técnico Entre em contato para dúvidas técnicas e suporte especializado.