tap-ixc

Biblioteca Python para sincronizar dados da API IXC Soft para PostgreSQL — com checkpointing por stage, retry, circuit breaker e observabilidade nativa.

Declare os endpoints que quer no clients.yml:

minha-empresa:
  system: ixc
  schema_name: public
  postgres_dsn: "postgresql://user:pass@host/db"
  api:
    base_url: "https://sua.ixcsoft.com.br/webservice/v1"
    token: "usuario:token"
  endpoints:
    - { name: clientes, api_endpoint: cliente,     strategy: full }
    - { name: titulos,  api_endpoint: fn_areceber, strategy: delta, pk_column: id }

E sincronize:

tap-ixc run minha-empresa
#   ✓ clientes: 12435 registros
#   ✓ titulos: 3120 registros

Por que não um script requests?

No mundo real	Script cru	tap-ixc
API caiu na página 800/1000	recomeça do zero	retry + circuit breaker por endpoint
Run diário pulou um dia	perde dados	cursor incremental retoma do último ponto
Falha no meio do INSERT	tabela em prod vazia	swap atômico (staging → COMMIT)
1 registro corrompido	derruba o batch	dead letter por linha
"Rodou? Quanto faltou?"	nenhuma pista	tabelas etl.* de observabilidade

Instalação

pip install git+https://github.com/arktnld/tap-ixc.git

Desenvolvimento local:

git clone https://github.com/arktnld/tap-ixc && cd tap-ixc
pip install -e ".[dev]"

Início rápido

Já instalado, do clients.yml ao primeiro sync:

# 1. criar a config a partir do exemplo e editar (base_url, token, postgres_dsn)
cp config/clients.yml.example config/clients.yml

# 2. apontar o banco de monitoramento e criar as tabelas
export ETL_MONITOR_DSN="postgresql://user:pass@localhost:5432/seu_db"
psql "$ETL_MONITOR_DSN" -f docs/schema.sql

# 3. validar (API + monitoramento) e rodar
tap-ixc check minha-empresa
tap-ixc run   minha-empresa       #   ✓ clientes: 12435 registros

Important

config/clients.yml não vem no repositório (apenas o .example). Se esquecer, a CLI avisa com o comando exato a rodar.

Para baixar só o que mudou, use strategy: delta no endpoint (cursor incremental por ultima_atualizacao). Cada run é idempotente e o load é atômico — basta agendar. Passo a passo completo, com explicações: Tutorial de 10 minutos.

Comandos

tap-ixc check minha-empresa                  # valida credenciais + monitoramento
tap-ixc discover minha-empresa               # lista streams e modos de sync
tap-ixc run minha-empresa                    # sincroniza todos os streams
tap-ixc run minha-empresa --stream clientes  # só um
tap-ixc run minha-empresa --from-checkpoint  # retoma do último checkpoint
tap-ixc list                                 # clientes configurados
tap-ixc status --client minha-empresa        # últimos runs

Note

run sai com código != 0 se qualquer stream falhar — seguro para cron e Airflow.

Uso via Python

Para uso programático, sem clients.yml:

from tap_ixc.tap import IXCTap, Destination
from tap_ixc.config.settings import ApiConfig

tap = IXCTap(ApiConfig(
    base_url="https://sua-instancia.ixcsoft.com.br/webservice/v1",
    token="usuario:token_aqui",
))

ok, err = tap.check_connection()
catalog = tap.discover()

destination = Destination(
    postgres_dsn="postgresql://user:pass@host/db",
    schema="public",
    duckdb_path="/tmp/staging.duckdb",
)
results = tap.sync(destination, catalog.select("clientes"))
# → [TapResult(stream="clientes", records_loaded=12435, status="success")]

Configuração

token, postgres_dsn e base_url aceitam valor literal ou ${VAR} (variável de ambiente). Pode misturar.

minha-empresa:
  system: ixc
  schema_name: public
  postgres_dsn: "postgresql://user:pass@localhost:5432/db"   # literal...
  duckdb_path: "/tmp/etl-staging/minha-empresa.duckdb"
  api:
    base_url: "https://minha-empresa.ixcsoft.com.br/webservice/v1"
    token: "${EMPRESA_API_TOKEN}"                            # ...ou via env
    max_retries: 3
    timeout_s: 60
  endpoints:
    - name: clientes
      api_endpoint: cliente
      strategy: full            # substitui a tabela toda
    - name: titulos
      api_endpoint: fn_areceber
      strategy: delta           # incremental por ultima_atualizacao
      pk_column: id

Tip

Secrets de produção ficam melhor em ${VAR} — não vão pro disco em texto puro. Para dev local, literal é mais prático.

Referência completa dos campos: Configuração.

Streams

Três já vêm registrados — clientes (cliente), contratos (cliente_contrato) e titulos (fn_areceber). Mas qualquer endpoint da API IXC vira um stream em dois passos:

# tap_ixc/streams/chamados.py
from tap_ixc.streams.base import Stream

class ChamadoStream(Stream):
    name = "chamados"                        # tabela destino
    api_endpoint = "su_oss_chamado"          # endpoint na API IXC
    replication_key = "ultima_atualizacao"   # None se não tiver modo delta

# tap_ixc/streams/__init__.py
STREAM_REGISTRY = {..., ChamadoStream.name: ChamadoStream}

A lib cuida do resto: paginação, retry, circuit breaker, staging DuckDB, load Postgres e checkpoint. Detalhes em Streams.

Validação e dead letter

Defina um schema pydantic no stream para ativar o stage VALIDATE: linhas reprovadas vão para etl.dead_letters (com o erro) e o batch nunca falha inteiro.

class ClienteSchema(BaseModel):
    id: int
    nome: str

class ClienteStream(Stream):
    name = "clientes"
    api_endpoint = "cliente"
    schema = ClienteSchema   # ativa o VALIDATE

Warning

O modo incremental só vê inserts/updates. Registros apagados na origem não são removidos do destino — rode strategy: full periodicamente para reconciliar.

Mais em Incremental e validação.

Agendamento (Airflow / cron)

from tap_ixc.runner import run

results = run("minha-empresa")
if any(r.status == "failed" for r in results):
    raise RuntimeError("sync falhou")        # Airflow detecta → retry

Exemplo de DAG em examples/airflow_dag.py; guia em Deploy.

Monitoramento

A lib grava em um schema PostgreSQL dedicado (pipeline_runs, pipeline_events, checkpoints, dead_letters). Inicialize com psql "$ETL_MONITOR_DSN" -f docs/schema.sql e consulte via tap-ixc status ou SQL. Ver Monitoramento.

Documentação

Documentação completa em arktnld.github.io/tap-ixc:

• Tutorial de 10 minutos
• Configuração e CLI
• API Python e referência da API
• Arquitetura
• Troubleshooting
• Changelog

Licença

MIT — veja LICENSE.