Saltar a contenido

Sincronización incremental

Estrategia: Incremental vs. Completado

La elección de la estrategia de sincronización impacta directamente el desempeño y el consumo del rate limit.

Estrategia Cuando use Ventaja Desventaje
Completado Primera sync, reconciliación semanal Garante consistencia total Consome más rate limit
Incremental Syncs regulares (horária, diaria) Rápido, consume menos rate limit Puede perder cambios en caso de fallo

Sincronización incremental

Sincroniza sólo registros modificados desde la última sincronización exitosa.

Flujo

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]

Implementación

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

Sincronización completa

Recomendada para la primera sync o reconciliación 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

Tratamiento de fallos

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)

Planificación recomendada

Frecuencia Estrategia Observación
Cada hora Incremental Mantener sistema en día
Diariamente (madrugada) Completado Reconciliación
Después de incidentes Completado Garantizar consistencia

Siguientes pasos