Limites e Quotas

Visão Geral

A API DataSnap implementa limites para garantir performance estável e uso justo dos recursos. Este guia detalha todos os limites aplicados e como trabalhar eficientemente dentro deles.

Os limites são aplicados por tenant (conta) e são resetados automaticamente dentro das janelas de tempo especificadas.

Limites por Endpoint

Upload de Arquivos

Limite	Valor	Observações
Tamanho máximo por arquivo	100 MB	Aplica-se a cada arquivo individual
Formato suportado	JSONL	Cada linha deve ser um JSON válido
Detecção de duplicatas	Automática	Via hash MD5
Taxa de requisições	Sem limite fixo	Sujeito a throttling automático

Listagem de Arquivos

Limite	Valor	Janela de Tempo
Requisições	100 requests	1 minuto
Itens por página	1-100	Por requisição

Consultas (Query)

Limite	Valor	Janela de Tempo
Requisições	50 requests	1 minuto
Registros por consulta	1-10.000	Por requisição
Timeout de consulta	30 segundos	Execução máxima

Processamento DataFlow

Limite	Valor	Janela de Tempo
Requisições	10 requests	5 minutos
Executores máximo	10	Por processamento
Tempo de execução	10-30 minutos	Varia conforme volume

Códigos de Resposta de Limite

429 - Too Many Requests

Retornado quando você excede os limites de taxa:

{
  "error": "Muitas requisições. Tente novamente em 60 segundos.",
  "retry_after": 60,
  "limit_type": "rate_limit",
  "current_usage": 51,
  "limit": 50
}

413 - Payload Too Large

Para arquivos que excedem o tamanho máximo:

{
  "error": "Arquivo muito grande. Limite máximo: 100MB",
  "file_size_mb": 150,
  "max_size_mb": 100
}

Boas Práticas

Gerenciamento de Taxa de Requisições

Implemente throttling em suas aplicações para evitar atingir os limites:

import time
from functools import wraps

def rate_limit(calls_per_minute=50):
    def decorator(func):
        last_called = [0.0]
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            left_to_wait = 60.0 / calls_per_minute - elapsed
            
            if left_to_wait > 0:
                time.sleep(left_to_wait)
                
            ret = func(*args, **kwargs)
            last_called[0] = time.time()
            return ret
            
        return wrapper
    return decorator

# Uso
@rate_limit(calls_per_minute=45)  # Margem de segurança
def fazer_consulta(query_data):
    return requests.post(url, headers=headers, json=query_data)

Upload Eficiente

Para múltiplos arquivos, otimize o processo:

def upload_multiplos_arquivos(schema_slug, lista_arquivos, headers):
    """Upload otimizado para múltiplos arquivos"""
    files = []
    try:
        # Preparar todos os arquivos
        for arquivo in lista_arquivos:
            files.append(('files', open(arquivo, 'rb')))
        
        # Upload em lote
        response = requests.post(
            f"https://api.datasnap.com.br/api/v1/schemas/{schema_slug}/files",
            headers=headers,
            files=files,
            timeout=300  # 5 minutos para uploads grandes
        )
        
        return response.json()
        
    finally:
        # Sempre fechar os arquivos
        for _, file_obj in files:
            file_obj.close()

# Dividir arquivos grandes em chunks menores
def dividir_arquivo_grande(arquivo_path, chunk_size_mb=80):
    """Divide arquivos grandes em chunks menores"""
    chunk_size = chunk_size_mb * 1024 * 1024
    chunks = []
    
    with open(arquivo_path, 'r') as f:
        chunk_lines = []
        current_size = 0
        
        for line in f:
            line_size = len(line.encode('utf-8'))
            
            if current_size + line_size > chunk_size and chunk_lines:
                # Salvar chunk atual
                chunk_file = f"{arquivo_path}.chunk_{len(chunks)}.jsonl"
                with open(chunk_file, 'w') as chunk_f:
                    chunk_f.writelines(chunk_lines)
                chunks.append(chunk_file)
                
                # Resetar para próximo chunk
                chunk_lines = [line]
                current_size = line_size
            else:
                chunk_lines.append(line)
                current_size += line_size
        
        # Último chunk
        if chunk_lines:
            chunk_file = f"{arquivo_path}.chunk_{len(chunks)}.jsonl"
            with open(chunk_file, 'w') as chunk_f:
                chunk_f.writelines(chunk_lines)
            chunks.append(chunk_file)
    
    return chunks

Consultas Otimizadas

Estruture suas consultas para melhor performance:

# ✅ Consulta otimizada
consulta_eficiente = {
    "select": ["id", "nome", "categoria"],  # Campos específicos
    "where": [
        {"field": "ativo", "op": "=", "value": True},
        {"field": "criado_em", "op": ">=", "value": "2024-01-01"}
    ],
    "order_by": [{"field": "id", "direction": "asc"}],
    "limit": 100,  # Limite razoável
    "format": "json_compact"  # Formato compacto para menos dados
}

# ❌ Consulta ineficiente
consulta_problematica = {
    "select": ["*"],  # Evite select * 
    "limit": 10000,   # Limite muito alto
    "format": "json"  # Formato verboso desnecessário
}

Paginação Eficiente

Use paginação baseada em cursor para melhor performance:

def buscar_todos_registros(schema_slug, headers, consulta_base):
    """Busca todos os registros usando paginação eficiente"""
    todos_registros = []
    page_token = None
    
    while True:
        consulta = consulta_base.copy()
        consulta["limit"] = 100  # Chunks menores
        
        if page_token:
            consulta["page_token"] = page_token
            
        try:
            response = requests.post(
                f"https://api.datasnap.com.br/api/v1/schemas/{schema_slug}/query",
                headers=headers,
                json=consulta,
                timeout=30
            )
            response.raise_for_status()
            
            dados = response.json()
            todos_registros.extend(dados.get("data", []))
            
            # Verificar se há mais páginas
            meta = dados.get("meta", {})
            if "next_page_token" not in meta:
                break
                
            page_token = meta["next_page_token"]
            
        except requests.exceptions.RequestException as e:
            print(f"Erro na paginação: {e}")
            break
    
    return todos_registros

Monitoramento de Uso

Implementando Logs Estruturados

import logging
import json
import time

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

def log_api_usage(endpoint, method, status_code, response_time, data_size=None):
    """Log estruturado para monitorar uso da API"""
    log_data = {
        "timestamp": time.time(),
        "endpoint": endpoint,
        "method": method,
        "status_code": status_code,
        "response_time_ms": response_time * 1000,
        "success": 200 <= status_code < 300,
    }
    
    if data_size:
        log_data["data_size_mb"] = data_size / (1024 * 1024)
    
    # Adicionar alertas para limites próximos
    if status_code == 429:
        log_data["alert"] = "RATE_LIMIT_EXCEEDED"
    elif response_time > 25:  # Próximo do timeout
        log_data["alert"] = "SLOW_RESPONSE"
    
    logger.info(json.dumps(log_data))

# Wrapper para requests com monitoramento
def fazer_requisicao_monitorada(method, url, **kwargs):
    """Wrapper que adiciona monitoramento automático"""
    start_time = time.time()
    
    try:
        response = requests.request(method, url, **kwargs)
        response_time = time.time() - start_time
        
        # Calcular tamanho dos dados se possível
        data_size = len(response.content) if response.content else 0
        
        log_api_usage(
            endpoint=url.split('/')[-1],
            method=method,
            status_code=response.status_code,
            response_time=response_time,
            data_size=data_size
        )
        
        return response
        
    except Exception as e:
        response_time = time.time() - start_time
        logger.error(f"Erro na requisição: {e}")
        log_api_usage(
            endpoint=url.split('/')[-1],
            method=method,
            status_code=0,
            response_time=response_time
        )
        raise

Estratégias de Otimização

Para Alta Frequência de Dados

Se você processa muitos dados regularmente:

Processamento em Lote

Configure o processamento automático baseado em volume para otimizar recursos:

# Em vez de processar arquivo por arquivo
def processamento_ineficiente():
    for arquivo in arquivos:
        upload_file(arquivo)
        process_files()  # Muitas chamadas de processamento

# Use processamento em lote
def processamento_otimizado():
    # Upload todos os arquivos primeiro
    for arquivo in arquivos:
        upload_file(arquivo)
    
    # Um único processamento para todos
    process_files()

Cache de Consultas

Implemente cache para consultas frequentes:

import time
from functools import lru_cache
import hashlib

class QueryCache:
    def __init__(self, ttl_seconds=300):  # 5 minutos
        self.cache = {}
        self.ttl = ttl_seconds
    
    def get_cache_key(self, query_data):
        """Gera chave única para a consulta"""
        query_str = json.dumps(query_data, sort_keys=True)
        return hashlib.md5(query_str.encode()).hexdigest()
    
    def get(self, query_data):
        key = self.get_cache_key(query_data)
        if key in self.cache:
            cached_data, timestamp = self.cache[key]
            if time.time() - timestamp < self.ttl:
                return cached_data
        return None
    
    def set(self, query_data, result):
        key = self.get_cache_key(query_data)
        self.cache[key] = (result, time.time())

# Uso do cache
cache = QueryCache(ttl_seconds=300)

def consulta_com_cache(query_data):
    # Tentar cache primeiro
    cached = cache.get(query_data)
    if cached:
        return cached
    
    # Fazer requisição se não estiver em cache
    result = fazer_consulta(query_data)
    cache.set(query_data, result)
    return result

Alertas e Monitoramento

Métricas Importantes

Monitore estas métricas em suas integrações:

Taxa de sucesso: % de requisições bem-sucedidas
Tempo de resposta médio: Performance da API
Taxa de erro por endpoint: Identifica problemas específicos
Uso de cota: Proximidade dos limites
Volume de dados processados: Crescimento ao longo do tempo

Configuração de Alertas

def verificar_saude_api():
    """Verificação periódica da saúde da API"""
    metrics = {
        "success_rate": 0.0,
        "avg_response_time": 0.0,
        "error_count": 0,
        "quota_usage": 0.0
    }
    
    # Calcular métricas dos últimos logs
    # ... lógica de cálculo ...
    
    # Alertas baseados em thresholds
    if metrics["success_rate"] < 0.95:
        enviar_alerta("Taxa de sucesso baixa", metrics)
    
    if metrics["avg_response_time"] > 20:
        enviar_alerta("API lenta", metrics)
    
    if metrics["quota_usage"] > 0.8:
        enviar_alerta("Quota próxima do limite", metrics)

def enviar_alerta(tipo, metrics):
    """Enviar alerta via webhook, email, etc."""
    alert_data = {
        "alert_type": tipo,
        "timestamp": time.time(),
        "metrics": metrics
    }
    # Implementar envio do alerta

Troubleshooting

Problemas Comuns

Error 429 - Rate Limit

Sintoma: Muitas requisições sendo rejeitadasSoluções:

Implementar rate limiting no cliente
Usar exponential backoff
Reduzir frequência de requisições
Usar processamento em lote quando possível

import random
import time

def exponential_backoff(func, max_retries=5):
    """Implementa retry com backoff exponencial"""
    for attempt in range(max_retries):
        try:
            return func()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:
                if attempt == max_retries - 1:
                    raise
                
                # Backoff exponencial com jitter
                delay = (2 ** attempt) + random.uniform(0, 1)
                time.sleep(delay)
            else:
                raise

Consultas Lentas

Sintoma: Timeout em consultas ou respostas muito demoradasSoluções:

Usar índices adequados nos filtros
Limitar o número de registros retornados
Usar formato json_compact
Otimizar cláusulas WHERE

# ✅ Consulta otimizada
query_rapida = {
    "select": ["id", "campo_indexado"],
    "where": [
        {"field": "campo_indexado", "op": "=", "value": "valor_especifico"}
    ],
    "limit": 100,
    "format": "json_compact"
}

Upload Failures

Sintoma: Falhas frequentes no upload de arquivosSoluções:

Verificar tamanho dos arquivos (máx 100MB)
Validar formato JSONL
Implementar retry para falhas temporárias
Dividir arquivos grandes

def validar_jsonl(arquivo_path):
    """Valida formato JSONL antes do upload"""
    try:
        with open(arquivo_path, 'r') as f:
            for i, linha in enumerate(f, 1):
                try:
                    json.loads(linha.strip())
                except json.JSONDecodeError as e:
                    return False, f"Linha {i}: {e}"
        return True, "Válido"
    except Exception as e:
        return False, str(e)

Planos e Limites Customizados

Para necessidades especiais de volume ou performance, entre em contato com nossa equipe:

Suporte Técnico

Para discutir limites customizados ou problemas de performance.

Vendas

Para planos empresariais com limites aumentados.

Os limites e quotas estão sujeitos a alterações. Sempre consulte a documentação mais recente e monitore seu uso de forma programática.

Começando

Personalização

Escrevendo conteúdo

Ferramentas de IA

Visão Geral

Limites por Endpoint

Upload de Arquivos

Listagem de Arquivos

Consultas (Query)

Processamento DataFlow

Códigos de Resposta de Limite

429 - Too Many Requests

413 - Payload Too Large

Boas Práticas

Gerenciamento de Taxa de Requisições

Upload Eficiente

Consultas Otimizadas

Paginação Eficiente

Monitoramento de Uso

Implementando Logs Estruturados

Estratégias de Otimização

Para Alta Frequência de Dados

Alertas e Monitoramento

Métricas Importantes

Configuração de Alertas

Troubleshooting

Problemas Comuns

Planos e Limites Customizados

Suporte Técnico

Vendas

Começando

Personalização

Escrevendo conteúdo

Ferramentas de IA

​Visão Geral

​Limites por Endpoint

​Upload de Arquivos

​Listagem de Arquivos

​Consultas (Query)

​Processamento DataFlow

​Códigos de Resposta de Limite

​429 - Too Many Requests

​413 - Payload Too Large

​Boas Práticas

​Gerenciamento de Taxa de Requisições

​Upload Eficiente

​Consultas Otimizadas

​Paginação Eficiente

​Monitoramento de Uso

​Implementando Logs Estruturados

​Estratégias de Otimização

​Para Alta Frequência de Dados

​Alertas e Monitoramento

​Métricas Importantes

​Configuração de Alertas

​Troubleshooting

​Problemas Comuns

​Planos e Limites Customizados

Suporte Técnico

Vendas

Visão Geral

Limites por Endpoint

Upload de Arquivos

Listagem de Arquivos

Consultas (Query)

Processamento DataFlow

Códigos de Resposta de Limite

429 - Too Many Requests

413 - Payload Too Large

Boas Práticas

Gerenciamento de Taxa de Requisições

Upload Eficiente

Consultas Otimizadas

Paginação Eficiente

Monitoramento de Uso

Implementando Logs Estruturados

Estratégias de Otimização

Para Alta Frequência de Dados

Alertas e Monitoramento

Métricas Importantes

Configuração de Alertas

Troubleshooting

Problemas Comuns

Planos e Limites Customizados