Skip to content

Incremental Synchronization

Strategy: Incremental vs. Complete

The choice of synchronization strategy directly impacts the performance and consumption of the rate limit.

Strategy When to use Advantage Disadvantage
Complete First sync, weekly reconciliation Ensures total consistency Consume more rate limit
Incremental Regular syncs (time, daily) Quick, use less rate limit You may miss changes in case of failure

Incremental Synchronization

Syncs only modified records since the last successful synchronization.

Flow

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]

Implementation

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

Full Synchronization

Recommended for first sync or periodic reconciliation:

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

Failure Treatment

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)
Frequency Strategy Observation
Every hour Incremental Keep system up to date
Daily (early) Complete Reconciliation
After incidents Complete Ensure consistency

Next Steps