Pular para conteúdo

Sincronização Incremental

Estratégia: Incremental vs. Completa

A escolha da estratégia de sincronização impacta diretamente a performance e o consumo do rate limit.

Estratégia Quando usar Vantagem Desvantagem
Completa Primeira sync, reconciliação semanal Garante consistência total Consome mais rate limit
Incremental Syncs regulares (horária, diária) Rápida, consome menos rate limit Pode perder mudanças em caso de falha

Sincronização Incremental

Sincroniza apenas registros modificados desde a última sincronização bem-sucedida.

Fluxo

flowchart TD
    A[Início] --> B[Ler timestamp da última sync]
    B --> C[Buscar registros modificados desde então]
    C --> D[Para cada registro]
    D --> E{Existe na Pontotel?}
    E -->|Não| F[Criar via POST]
    E -->|Sim| G[Atualizar via PATCH]
    F --> H[Salvar resultado]
    G --> H
    H --> I[Salvar novo timestamp]
    I --> J[Fim]

Implementação

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
import json
from datetime import datetime, timezone
from pathlib import Path

SYNC_STATE_FILE = Path(".sync_state.json")

def carregar_estado() -> dict:
    """Carrega o estado da última sincronização"""
    if SYNC_STATE_FILE.exists():
        return json.loads(SYNC_STATE_FILE.read_text())
    return {"ultima_sync": None}

def salvar_estado(timestamp: str):
    """Salva o timestamp da sincronização bem-sucedida"""
    SYNC_STATE_FILE.write_text(json.dumps({"ultima_sync": timestamp}))

def sincronizar_incremental(auth):
    estado = carregar_estado()
    ultima_sync = estado["ultima_sync"]

    inicio = datetime.now(timezone.utc).isoformat()

    print(f"📅 Sincronizando desde: {ultima_sync or 'início'}")

    # Sincronizar empregadores
    resultado_emp = sincronizar_empregadores_desde(ultima_sync, auth)

    # Sincronizar empregados
    resultado_func = sincronizar_empregados_desde(ultima_sync, auth)

    # Salvar novo estado somente se tudo deu certo
    salvar_estado(inicio)

    print(f"✅ Sync concluída!")
    print(f"   Empregadores: {resultado_emp}")
    print(f"   Empregados: {resultado_func}")

Sincronização Completa

Recomendada para a primeira sync ou reconciliação periódica:

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def sincronizar_completa(auth):
    """Sincronização completa — compara estado atual com Pontotel"""
    headers = auth.get_headers()

    # 1. Carregar todos os registros do seu sistema
    seus_empregados = seu_sistema.listar_todos_empregados()

    # 2. Carregar todos os registros da Pontotel
    empregados_pontotel = listar_todos_da_api(
        "https://apis.pontotel.com.br/pontotel/api/v4/empregados/",
        headers
    )
    pontotel_por_cpf = {e["cpf"]: e for e in empregados_pontotel}

    resultado = {"criados": 0, "atualizados": 0, "desativados": 0}

    for func in seus_empregados:
        cpf = func["cpf"]
        if cpf not in pontotel_por_cpf:
            criar_empregado(func, headers)
            resultado["criados"] += 1
        elif dados_diferentes(pontotel_por_cpf[cpf], func):
            atualizar_empregado(pontotel_por_cpf[cpf]["id"], func, headers)
            resultado["atualizados"] += 1

    return resultado

Tratamento de Falhas

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
import time

def sync_com_retry(auth, max_tentativas=3):
    """Sincronização com retry automático em caso de falha"""
    for tentativa in range(1, max_tentativas + 1):
        try:
            return sincronizar_incremental(auth)
        except RateLimitError as e:
            wait = e.retry_after
            print(f"⏳ Rate limit: aguardando {wait}s...")
            time.sleep(wait)
        except APIError as e:
            if tentativa == max_tentativas:
                raise
            wait = 2 ** tentativa  # Backoff exponencial: 2s, 4s, 8s
            print(f"❌ Tentativa {tentativa} falhou. Aguardando {wait}s...")
            time.sleep(wait)

Agendamento Recomendado

Frequência Estratégia Observação
A cada hora Incremental Manter sistema em dia
Diariamente (madrugada) Completa Reconciliação
Após incidentes Completa Garantir consistência

Próximos Passos