Ir para o conteúdo

Referência da API

Gerada automaticamente do código (docstrings e assinaturas).

Tap

tap_ixc.tap.IXCTap

Tap IXC — sincroniza streams de uma API IXC para Postgres.

Streams disponíveis por padrão: clientes, contratos, titulos.

Source code in tap_ixc/tap.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
class IXCTap:
    """
    Tap IXC — sincroniza streams de uma API IXC para Postgres.

    Streams disponíveis por padrão: clientes, contratos, titulos.
    """

    # Fonte única: o registry. Adicionar stream lá reflete aqui automaticamente.
    STREAMS: list[type[Stream]] = list(STREAM_REGISTRY.values())

    def __init__(self, config: ApiConfig) -> None:
        self._config = config
        self._client = IXCClient(
            base_url=config.base_url,
            token=config.token,
            max_retries=config.max_retries,
            timeout_s=config.timeout_s,
            backoff_factor=config.backoff_factor,
            wait_jitter=config.wait_jitter,
            session_renewal_every=config.session_renewal_every,
            rate_limit_sleep=config.rate_limit_sleep,
        )

    def check_connection(self) -> tuple[bool, str | None]:
        """
        Verifica se as credenciais são válidas antes de rodar.
        Retorna (True, None) se ok, (False, mensagem) se falhar.
        """
        return self._client.check_connection()

    def discover(self) -> Catalog:
        """
        Retorna o Catalog com todos os streams disponíveis.
        sync_mode padrão: INCREMENTAL se o stream tem replication_key, FULL caso contrário.
        """
        entries = [
            CatalogEntry(
                stream=cls,
                destination_table=cls.name,
                sync_mode=SyncMode.INCREMENTAL if cls.replication_key else SyncMode.FULL,
                pk_column=cls.primary_keys[0],
            )
            for cls in self.STREAMS
        ]
        return Catalog(entries)

    def sync(
        self,
        destination: Destination,
        catalog: Catalog | None = None,
        from_checkpoint: bool = False,
        client_id: str = "default",
    ) -> list[TapResult]:
        """
        Sincroniza streams para o destino.

        catalog=None → sincroniza todos os streams disponíveis.
        client_id  → identificador gravado nos logs de observabilidade.
        """
        settings = Settings()
        checkpoint = Checkpoint(settings.monitor_dsn, schema=settings.monitor_schema)
        events = EventStore(settings.monitor_dsn, schema=settings.monitor_schema)

        active_catalog = catalog if catalog is not None else self.discover()
        results = []

        for entry in active_catalog:
            log.info("tap.stream_start", stream=entry.stream.name, mode=entry.sync_mode)
            result = self._sync_entry(
                entry, destination, checkpoint, events, from_checkpoint, client_id
            )
            results.append(result)
            log.info("tap.stream_done", stream=entry.stream.name, status=result.status)

        return results

    def _sync_entry(
        self,
        entry: CatalogEntry,
        destination: Destination,
        checkpoint: Checkpoint,
        events: EventStore,
        from_checkpoint: bool,
        client_id: str,
    ) -> TapResult:
        stream = entry.stream()
        rep_key = stream.replication_key

        # Incremental: retoma do último cursor salvo no checkpoint, não de
        # "ontem meia-noite" hardcoded — assim pular um run não perde dados.
        since: str | None = None
        if entry.sync_mode == SyncMode.INCREMENTAL and rep_key:
            last_cp = checkpoint.get_last(client_id, entry.destination_table)
            if last_cp and last_cp.get("metadata"):
                since = last_cp["metadata"].get("replication_key_value")

        staging = StagingLoader(
            duckdb_path=destination.duckdb_path,
            table=entry.destination_table,
            fields=entry.selected_fields,
            transform_sql=entry.transform_sql,
        )
        pg_loader = PostgresLoader(
            duckdb_path=destination.duckdb_path,
            pg_dsn=destination.postgres_dsn,
            schema=destination.schema,
            table=entry.destination_table,
            strategy=entry.sync_mode.value,
            pk_column=entry.pk_column,
        )

        def extract_stage(ctx: PipelineContext) -> None:
            cursor: dict[str, str | None] = {"v": since}

            def tracked(it):
                for rec in it:
                    if rep_key:
                        cursor["v"] = _advance_cursor(cursor["v"], rec.get(rep_key))
                    yield rec

            records = stream.get_records(
                self._client, entry.sync_mode, since=since, page_size=entry.page_size
            )
            ndjson_path, total = staging.load(tracked(records))
            ctx.set("data_path", ndjson_path)
            ctx.set("records_extracted", total)
            if rep_key and cursor["v"] is not None:
                ctx.set("new_cursor", cursor["v"])

        def validate_stage(ctx: PipelineContext) -> None:
            # Só roda quando o stream define schema (senão nem é registrado abaixo).
            # Extração vazia: nem tabela de staging existe — nada a validar.
            if ctx.get("records_extracted") == 0:
                ctx.set("records_valid", 0)
                return
            rows = staging.read_all()
            valid, dead = validate_batch(rows, schema=stream.schema)
            if dead:
                events.write_dead_letters(ctx.get("run_id"), stream.name, dead)
                ctx.set("records_dead", len(dead))
            staging.replace(valid)
            ctx.set("records_valid", len(valid))

        def load_stage(ctx: PipelineContext) -> None:
            # Extração vazia (ex: incremental sem mudanças): nada a carregar.
            # Não dropa/recria a tabela à toa — vira no-op de sucesso.
            if ctx.get("records_extracted") == 0:
                ctx.set("records_loaded", 0)
                return
            ctx.set("records_loaded", pg_loader.load())

        def verify_stage(ctx: PipelineContext) -> None:
            loaded = ctx.get("records_loaded", 0)
            extracted = ctx.get("records_extracted")
            # EXTRACT pulado (resume via --from-checkpoint): não há contagem
            # desta sessão para comparar — confia no checkpoint anterior.
            if extracted is None:
                return
            if extracted > 0 and loaded == 0:
                raise RuntimeError(
                    f"Verificação falhou: {extracted} extraídos, 0 carregados"
                )
            # Se VALIDATE rodou, o esperado é o nº de válidas (dead letters saíram).
            expected = ctx.get("records_valid", extracted)
            if loaded != expected:
                raise RuntimeError(
                    f"Contagem divergente: {expected} esperados, {loaded} carregados"
                )

        pipeline = PipelineRun(
            client=client_id,
            system="ixc",
            pipeline=entry.destination_table,
            checkpoint=checkpoint,
            events=events,
            from_checkpoint=from_checkpoint,
        )

        try:
            stages = {
                Stage.EXTRACT: extract_stage,
                Stage.LOAD: load_stage,
                Stage.VERIFY: verify_stage,
            }
            if stream.schema is not None:
                stages[Stage.VALIDATE] = validate_stage

            ctx = pipeline.execute(stages)
            # Avança o cursor SÓ após EXTRACT+LOAD+VERIFY ok. Se LOAD falhar,
            # o cursor não anda e o próximo run rebusca a mesma janela.
            new_cursor = ctx.get("new_cursor")
            if new_cursor is not None:
                checkpoint.mark_done(
                    client_id,
                    entry.destination_table,
                    Stage.VERIFY.value,
                    data_path=ctx.get("data_path"),
                    metadata={"replication_key_value": new_cursor},
                )
            return TapResult(
                stream=stream.name,
                records_extracted=ctx.get("records_extracted", 0),
                records_loaded=ctx.get("records_loaded", 0),
                status="success",
            )
        except Exception as exc:
            # Conta parcial do ctx do pipeline (EXTRACT pode ter completado antes da falha).
            failed = pipeline.ctx
            return TapResult(
                stream=stream.name,
                records_extracted=failed.get("records_extracted", 0),
                records_loaded=failed.get("records_loaded", 0),
                status="failed",
                error=redact(str(exc)),
            )

check_connection()

Verifica se as credenciais são válidas antes de rodar. Retorna (True, None) se ok, (False, mensagem) se falhar.

Source code in tap_ixc/tap.py 
103
104
105
106
107
108
def check_connection(self) -> tuple[bool, str | None]:
    """
    Verifica se as credenciais são válidas antes de rodar.
    Retorna (True, None) se ok, (False, mensagem) se falhar.
    """
    return self._client.check_connection()

discover()

Retorna o Catalog com todos os streams disponíveis. sync_mode padrão: INCREMENTAL se o stream tem replication_key, FULL caso contrário.

Source code in tap_ixc/tap.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def discover(self) -> Catalog:
    """
    Retorna o Catalog com todos os streams disponíveis.
    sync_mode padrão: INCREMENTAL se o stream tem replication_key, FULL caso contrário.
    """
    entries = [
        CatalogEntry(
            stream=cls,
            destination_table=cls.name,
            sync_mode=SyncMode.INCREMENTAL if cls.replication_key else SyncMode.FULL,
            pk_column=cls.primary_keys[0],
        )
        for cls in self.STREAMS
    ]
    return Catalog(entries)

sync(destination, catalog=None, from_checkpoint=False, client_id='default')

Sincroniza streams para o destino.

catalog=None → sincroniza todos os streams disponíveis. client_id → identificador gravado nos logs de observabilidade.

Source code in tap_ixc/tap.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def sync(
    self,
    destination: Destination,
    catalog: Catalog | None = None,
    from_checkpoint: bool = False,
    client_id: str = "default",
) -> list[TapResult]:
    """
    Sincroniza streams para o destino.

    catalog=None → sincroniza todos os streams disponíveis.
    client_id  → identificador gravado nos logs de observabilidade.
    """
    settings = Settings()
    checkpoint = Checkpoint(settings.monitor_dsn, schema=settings.monitor_schema)
    events = EventStore(settings.monitor_dsn, schema=settings.monitor_schema)

    active_catalog = catalog if catalog is not None else self.discover()
    results = []

    for entry in active_catalog:
        log.info("tap.stream_start", stream=entry.stream.name, mode=entry.sync_mode)
        result = self._sync_entry(
            entry, destination, checkpoint, events, from_checkpoint, client_id
        )
        results.append(result)
        log.info("tap.stream_done", stream=entry.stream.name, status=result.status)

    return results

tap_ixc.tap.Destination dataclass

Configuração do destino Postgres.

Source code in tap_ixc/tap.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
@dataclass
class Destination:
    """Configuração do destino Postgres."""
    postgres_dsn: str
    schema: str
    duckdb_path: str

    def __post_init__(self) -> None:
        if not self.postgres_dsn:
            raise ValueError("Destination.postgres_dsn não pode ser vazio")
        if not self.duckdb_path:
            raise ValueError("Destination.duckdb_path não pode ser vazio")
        validate_identifier(self.schema, "schema do destino")

tap_ixc.tap.TapResult dataclass

Resultado da sincronização de um stream.

Source code in tap_ixc/tap.py 
56
57
58
59
60
61
62
63
@dataclass
class TapResult:
    """Resultado da sincronização de um stream."""
    stream: str
    records_extracted: int
    records_loaded: int
    status: str            # "success" | "failed"
    error: str | None = None

tap_ixc.catalog.Catalog dataclass

Conjunto de streams a sincronizar.

Source code in tap_ixc/catalog.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@dataclass
class Catalog:
    """Conjunto de streams a sincronizar."""
    entries: list[CatalogEntry] = field(default_factory=list)

    def select(self, *names: str) -> "Catalog":
        """Retorna novo Catalog apenas com os streams indicados."""
        name_set = set(names)
        selected = [e for e in self.entries if e.stream.name in name_set]
        missing = name_set - {e.stream.name for e in selected}
        if missing:
            available = [e.stream.name for e in self.entries]
            raise ValueError(
                f"Streams não encontrados: {sorted(missing)}. "
                f"Disponíveis: {available}"
            )
        return Catalog(selected)

    def __iter__(self) -> Iterator[CatalogEntry]:
        return iter(self.entries)

    def __len__(self) -> int:
        return len(self.entries)

    def __repr__(self) -> str:
        names = [e.stream.name for e in self.entries]
        return f"Catalog(streams={names})"

select(*names)

Retorna novo Catalog apenas com os streams indicados.

Source code in tap_ixc/catalog.py 
40
41
42
43
44
45
46
47
48
49
50
51
def select(self, *names: str) -> "Catalog":
    """Retorna novo Catalog apenas com os streams indicados."""
    name_set = set(names)
    selected = [e for e in self.entries if e.stream.name in name_set]
    missing = name_set - {e.stream.name for e in selected}
    if missing:
        available = [e.stream.name for e in self.entries]
        raise ValueError(
            f"Streams não encontrados: {sorted(missing)}. "
            f"Disponíveis: {available}"
        )
    return Catalog(selected)

tap_ixc.catalog.CatalogEntry dataclass

Configura como um stream é sincronizado para o destino.

Source code in tap_ixc/catalog.py 
23
24
25
26
27
28
29
30
31
32
@dataclass
class CatalogEntry:
    """Configura como um stream é sincronizado para o destino."""
    stream: type["Stream"]
    destination_table: str
    sync_mode: SyncMode = SyncMode.FULL
    selected_fields: list[str] | None = None  # None = todos
    pk_column: str = "id"
    transform_sql: str | None = None           # sobrescreve selected_fields
    page_size: int | None = None               # None = usa default do IXCClient (5000)

tap_ixc.catalog.SyncMode

Bases: str, Enum

Source code in tap_ixc/catalog.py 
18
19
20
class SyncMode(str, Enum):
    FULL = "full"
    INCREMENTAL = "incremental"

Stream

tap_ixc.streams.base.Stream

Representa um stream de dados da API IXC.

Para criar um novo stream, basta herdar e definir os atributos:

class MeuStream(Stream):
    name = "meu_stream"
    api_endpoint = "meu_endpoint"
    replication_key = "ultima_atualizacao"  # para modo incremental
Source code in tap_ixc/streams/base.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class Stream:
    """
    Representa um stream de dados da API IXC.

    Para criar um novo stream, basta herdar e definir os atributos:

        class MeuStream(Stream):
            name = "meu_stream"
            api_endpoint = "meu_endpoint"
            replication_key = "ultima_atualizacao"  # para modo incremental
    """

    name: str
    api_endpoint: str
    primary_keys: list[str] = ["id"]
    replication_key: str | None = None  # campo para sync incremental
    # Schema pydantic opcional. Se definido, o stage VALIDATE valida cada row
    # e manda as reprovadas para dead letter (etl.dead_letters). None = só sanitiza.
    schema: type["BaseModel"] | None = None

    def get_records(
        self,
        client: "IXCClient",
        sync_mode: "SyncMode",
        since: str | None = None,
        page_size: int | None = None,
    ) -> Iterator[dict[str, Any]]:
        """Retorna registros da API. Delega paginação ao IXCClient."""
        from tap_ixc.catalog import SyncMode as SM

        strategy = (
            "delta"
            if sync_mode == SM.INCREMENTAL and self.replication_key
            else "full"
        )
        yield from client.paginate(
            endpoint=self.api_endpoint,
            strategy=strategy,
            incremental_since=since,
            page_size=page_size,
        )

get_records(client, sync_mode, since=None, page_size=None)

Retorna registros da API. Delega paginação ao IXCClient.

Source code in tap_ixc/streams/base.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def get_records(
    self,
    client: "IXCClient",
    sync_mode: "SyncMode",
    since: str | None = None,
    page_size: int | None = None,
) -> Iterator[dict[str, Any]]:
    """Retorna registros da API. Delega paginação ao IXCClient."""
    from tap_ixc.catalog import SyncMode as SM

    strategy = (
        "delta"
        if sync_mode == SM.INCREMENTAL and self.replication_key
        else "full"
    )
    yield from client.paginate(
        endpoint=self.api_endpoint,
        strategy=strategy,
        incremental_since=since,
        page_size=page_size,
    )

Cliente HTTP

tap_ixc.extractors.api.IXCClient

HTTP client para APIs IXC.

A API usa GET com body JSON (não query string). Circuit breaker por endpoint — um endpoint falhando não afeta os outros.

Source code in tap_ixc/extractors/api.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
class IXCClient:
    """
    HTTP client para APIs IXC.

    A API usa GET com body JSON (não query string).
    Circuit breaker por endpoint — um endpoint falhando não afeta os outros.
    """

    def __init__(
        self,
        base_url: str,
        token: str,
        page_size: int = 5000,
        timeout_s: int = 60,
        max_retries: int = 5,
        backoff_factor: float = 0.5,
        wait_jitter: float = 1.0,
        session_renewal_every: int = 0,
        rate_limit_sleep: float = 0.0,
    ) -> None:
        encoded = base64.b64encode(token.encode()).decode()
        self._base_url = base_url.rstrip("/")
        self._auth = f"Basic {encoded}"
        self._page_size = page_size
        self._timeout_s = timeout_s
        self._max_retries = max_retries
        self._backoff_factor = backoff_factor
        self._wait_jitter = wait_jitter
        self._session_renewal_every = session_renewal_every
        self._rate_limit_sleep = rate_limit_sleep

    def _headers(self) -> dict[str, str]:
        return {
            "Authorization": self._auth,
            "Content-Type": "application/json",
            "ixcsoft": "listar",
            "Accept": "application/json",
        }

    def _build_params(
        self,
        endpoint: str,
        page: int,
        strategy: str,
        incremental_since: str | None,
        rp: int | None = None,
    ) -> dict[str, Any]:
        base: dict[str, Any] = {
            "page": str(page),
            "rp": str(rp if rp is not None else self._page_size),
            "sortname": f"{endpoint}.id",
            "sortorder": "desc",
        }
        if strategy == "delta":
            since = incremental_since or _midnight_yesterday_sp()
            base.update({
                "qtype": f"{endpoint}.ultima_atualizacao",
                "query": since,
                "oper": ">=",
            })
        else:
            base.update({
                "qtype": f"{endpoint}.id",
                "query": "0",
                "oper": ">=",
            })
        return base

    def _fetch_page(
        self,
        client: httpx.Client,
        endpoint: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        breaker = get_circuit_breaker(endpoint)
        url = f"{self._base_url}/{endpoint}"
        payload = json.dumps(params)

        @stamina.retry(
            on=(
                httpx.TimeoutException,      # ConnectTimeout, ReadTimeout, WriteTimeout, PoolTimeout
                httpx.NetworkError,          # ConnectError (SSL), ReadError, WriteError
                httpx.RemoteProtocolError,   # resposta malformada, JSON inválido (wrappado abaixo)
                httpx.HTTPStatusError,       # 5xx (4xx filtrados antes de raise_for_status)
            ),
            attempts=self._max_retries,
            wait_initial=self._backoff_factor,
            wait_max=self._backoff_factor * 60,
            wait_jitter=self._wait_jitter,
        )
        def _do() -> dict[str, Any]:
            resp = breaker.call(
                client.request,
                method="GET",
                url=url,
                content=payload,
                headers=self._headers(),
                timeout=self._timeout_s,
            )
            # Erros permanentes — qualquer 4xx (exceto 429) não sofre retry
            if 400 <= resp.status_code < 500 and resp.status_code != 429:
                raise _PermanentHTTPError(
                    resp.status_code,
                    f"HTTP {resp.status_code}: {resp.text[:200]}",
                )
            # 429 — rate limit: respeitar Retry-After, com teto para não travar a
            # aplicação se o servidor mandar um valor absurdo (ex: 30 dias).
            if resp.status_code == 429:
                retry_after = min(
                    float(resp.headers.get("Retry-After", self._backoff_factor * 4)),
                    _MAX_RETRY_AFTER_S,
                )
                time.sleep(retry_after)
                raise httpx.ReadTimeout(f"Rate limited (429), aguardou {retry_after:.1f}s")
            # 5xx e outros 4xx → raise_for_status → HTTPStatusError → retry
            resp.raise_for_status()
            # JSON inválido → RemoteProtocolError → retry
            try:
                data = resp.json()
            except json.JSONDecodeError as exc:
                raise httpx.RemoteProtocolError(
                    f"JSON inválido (endpoint={endpoint}): {exc}"
                ) from exc
            # Erro de negócio da API → permanente, sem retry
            if isinstance(data, dict) and data.get("type") == "error":
                raise _PermanentHTTPError(0, f"API error: {data.get('message')}")
            return data  # type: ignore[return-value]

        return _do()

    def check_connection(self) -> tuple[bool, str | None]:
        """
        Verifica se as credenciais são válidas.
        Retorna (True, None) se ok, (False, mensagem_erro) se falhar.
        """
        try:
            with httpx.Client() as client:
                params = {
                    "page": "1", "rp": "1",
                    "qtype": "cliente.id", "query": "0", "oper": ">=",
                    "sortname": "cliente.id", "sortorder": "desc",
                }
                self._fetch_page(client, "cliente", params)
            return True, None
        except Exception as exc:
            return False, str(exc)

    def paginate(
        self,
        endpoint: str,
        strategy: str = "full",
        incremental_since: str | None = None,
        page_size: int | None = None,
    ) -> Iterator[dict[str, Any]]:
        """Pagina um endpoint IXC, yielding um registro por vez."""
        page = 1
        total_fetched = 0
        rp = page_size if page_size is not None else self._page_size
        start_time = time.monotonic()

        client = httpx.Client()
        try:
            while True:
                params = self._build_params(endpoint, page, strategy, incremental_since, rp)

                # SSL reconnection: recriar sessão ao capturar ConnectError
                try:
                    response = self._fetch_page(client, endpoint, params)
                except httpx.ConnectError:
                    log.warning("client.ssl_reconnect", endpoint=endpoint, page=page)
                    client.close()
                    client = httpx.Client()
                    response = self._fetch_page(client, endpoint, params)

                # Renovação preventiva de sessão a cada N páginas
                if self._session_renewal_every and page % self._session_renewal_every == 0:
                    log.info("client.session_renewed", endpoint=endpoint, page=page)
                    client.close()
                    client = httpx.Client()

                records: list[dict[str, Any]] = response.get("registros", []) or []
                total_raw = response.get("total", "")
                # Parse robusto: aceita "123"/"123.0", rejeita vazio/negativo/lixo.
                try:
                    total_api: int | None = int(float(total_raw))
                    if total_api < 0:
                        total_api = None
                except (ValueError, TypeError):
                    total_api = None

                if not records:
                    break

                for rec in records:
                    yield rec

                total_fetched += len(records)

                # ETA logging
                elapsed = time.monotonic() - start_time
                pct = (total_fetched / total_api * 100) if total_api else None
                eta_s = (
                    elapsed / total_fetched * (total_api - total_fetched)
                    if (total_api and total_fetched > 0 and total_fetched < total_api)
                    else None
                )
                total_pages = math.ceil(total_api / rp) if total_api else None
                log.info(
                    "client.page_done",
                    endpoint=endpoint,
                    page=f"{page}/{total_pages}" if total_pages else page,
                    fetched=total_fetched,
                    total=total_api,
                    pct=round(pct, 1) if pct is not None else None,
                    eta_s=round(eta_s) if eta_s is not None else None,
                )

                if total_api is not None and total_fetched >= total_api:
                    break

                # Trava de segurança: nunca paginar indefinidamente.
                if page >= _MAX_PAGES:
                    log.warning("client.max_pages_reached", endpoint=endpoint, pages=page)
                    break

                # Rate limiting entre páginas
                if self._rate_limit_sleep > 0:
                    time.sleep(self._rate_limit_sleep)

                page += 1
        finally:
            client.close()

        log.info("client.done", endpoint=endpoint, total=total_fetched)

check_connection()

Verifica se as credenciais são válidas. Retorna (True, None) se ok, (False, mensagem_erro) se falhar.

Source code in tap_ixc/extractors/api.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def check_connection(self) -> tuple[bool, str | None]:
    """
    Verifica se as credenciais são válidas.
    Retorna (True, None) se ok, (False, mensagem_erro) se falhar.
    """
    try:
        with httpx.Client() as client:
            params = {
                "page": "1", "rp": "1",
                "qtype": "cliente.id", "query": "0", "oper": ">=",
                "sortname": "cliente.id", "sortorder": "desc",
            }
            self._fetch_page(client, "cliente", params)
        return True, None
    except Exception as exc:
        return False, str(exc)

paginate(endpoint, strategy='full', incremental_since=None, page_size=None)

Pagina um endpoint IXC, yielding um registro por vez.

Source code in tap_ixc/extractors/api.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
def paginate(
    self,
    endpoint: str,
    strategy: str = "full",
    incremental_since: str | None = None,
    page_size: int | None = None,
) -> Iterator[dict[str, Any]]:
    """Pagina um endpoint IXC, yielding um registro por vez."""
    page = 1
    total_fetched = 0
    rp = page_size if page_size is not None else self._page_size
    start_time = time.monotonic()

    client = httpx.Client()
    try:
        while True:
            params = self._build_params(endpoint, page, strategy, incremental_since, rp)

            # SSL reconnection: recriar sessão ao capturar ConnectError
            try:
                response = self._fetch_page(client, endpoint, params)
            except httpx.ConnectError:
                log.warning("client.ssl_reconnect", endpoint=endpoint, page=page)
                client.close()
                client = httpx.Client()
                response = self._fetch_page(client, endpoint, params)

            # Renovação preventiva de sessão a cada N páginas
            if self._session_renewal_every and page % self._session_renewal_every == 0:
                log.info("client.session_renewed", endpoint=endpoint, page=page)
                client.close()
                client = httpx.Client()

            records: list[dict[str, Any]] = response.get("registros", []) or []
            total_raw = response.get("total", "")
            # Parse robusto: aceita "123"/"123.0", rejeita vazio/negativo/lixo.
            try:
                total_api: int | None = int(float(total_raw))
                if total_api < 0:
                    total_api = None
            except (ValueError, TypeError):
                total_api = None

            if not records:
                break

            for rec in records:
                yield rec

            total_fetched += len(records)

            # ETA logging
            elapsed = time.monotonic() - start_time
            pct = (total_fetched / total_api * 100) if total_api else None
            eta_s = (
                elapsed / total_fetched * (total_api - total_fetched)
                if (total_api and total_fetched > 0 and total_fetched < total_api)
                else None
            )
            total_pages = math.ceil(total_api / rp) if total_api else None
            log.info(
                "client.page_done",
                endpoint=endpoint,
                page=f"{page}/{total_pages}" if total_pages else page,
                fetched=total_fetched,
                total=total_api,
                pct=round(pct, 1) if pct is not None else None,
                eta_s=round(eta_s) if eta_s is not None else None,
            )

            if total_api is not None and total_fetched >= total_api:
                break

            # Trava de segurança: nunca paginar indefinidamente.
            if page >= _MAX_PAGES:
                log.warning("client.max_pages_reached", endpoint=endpoint, pages=page)
                break

            # Rate limiting entre páginas
            if self._rate_limit_sleep > 0:
                time.sleep(self._rate_limit_sleep)

            page += 1
    finally:
        client.close()

    log.info("client.done", endpoint=endpoint, total=total_fetched)

Configuração

tap_ixc.config.settings.ApiConfig

Bases: BaseModel

Source code in tap_ixc/config/settings.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class ApiConfig(BaseModel):
    base_url: str
    token: str
    max_retries: int = 3
    backoff_factor: float = 0.5
    timeout_s: int = 60
    wait_jitter: float = 1.0           # jitter no backoff (anti-thundering-herd)
    session_renewal_every: int = 0     # recriar sessão a cada N páginas (0 = desligado)
    rate_limit_sleep: float = 0.0      # pausa entre páginas em segundos (0 = desligado)

    @field_validator("base_url")
    @classmethod
    def _valid_base_url(cls, v: str) -> str:
        # ${VAR} ainda não-expandido passa (expansão ocorre depois, em ClientConfig).
        if "${" in v:
            return v
        parsed = urlparse(v)
        if parsed.scheme not in ("http", "https") or not parsed.netloc:
            raise ValueError(
                f"base_url inválida: {v!r} — esperado http(s)://host/caminho"
            )
        return v

tap_ixc.config.settings.ClientConfig

Bases: BaseModel

Source code in tap_ixc/config/settings.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class ClientConfig(BaseModel):
    client: str
    system: str
    schema_name: str
    postgres_dsn: str
    api: ApiConfig
    endpoints: list[EndpointConfig] = []
    webhook_url: str | None = None
    duckdb_path: str = "/tmp/etl-staging/{client}.duckdb"

    @model_validator(mode="after")
    def _resolve_env(self) -> "ClientConfig":
        """Expande ${VAR} em campos de string."""
        self.postgres_dsn = _expand(self.postgres_dsn)
        self.api.base_url = _expand(self.api.base_url)
        self.api.token = _expand(self.api.token)
        if self.webhook_url:
            self.webhook_url = _expand(self.webhook_url)
        return self

    def duckdb_resolved(self) -> str:
        return self.duckdb_path.format(client=self.client)

tap_ixc.config.settings.EndpointConfig

Bases: BaseModel

Source code in tap_ixc/config/settings.py 
39
40
41
42
43
44
45
46
47
48
class EndpointConfig(BaseModel):
    name: str           # nome da tabela destino (ex: clientes)
    api_endpoint: str   # nome do endpoint na API (ex: cliente)
    strategy: Literal["full", "delta"] = "full"
    pk_column: str = "id"
    page_size: int = 5000
    # Lista de campos a manter — None = SELECT * (todos os campos)
    fields: list[str] | None = None
    # SQL de transformação customizado — sobrescreve `fields` se definido
    transform_sql: str | None = None

tap_ixc.config.settings.get_client(name, path=None)

Source code in tap_ixc/config/settings.py 
154
155
156
157
158
159
160
161
162
163
def get_client(name: str, path: Path | None = None) -> ClientConfig:
    clients = load_clients(path)
    if name not in clients:
        raise ValueError(
            f"Cliente '{name}' não encontrado em clients.yml. "
            f"Disponíveis: {sorted(clients)}"
        )
    cfg = clients[name]
    _assert_resolved(cfg)
    return cfg

tap_ixc.config.settings.load_clients(path=None)

Source code in tap_ixc/config/settings.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def load_clients(path: Path | None = None) -> dict[str, ClientConfig]:
    settings = Settings()
    yml_path = path or settings.clients_yml
    if not yml_path.exists():
        raise FileNotFoundError(
            f"Config de clientes não encontrada: {yml_path}\n"
            f"Crie a partir do exemplo:\n"
            f"    cp config/clients.yml.example config/clients.yml\n"
            f"e edite com base_url, token e postgres_dsn do seu cliente."
        )
    try:
        raw = yaml.safe_load(yml_path.read_text())
    except yaml.YAMLError as exc:
        raise ValueError(f"clients.yml com sintaxe YAML inválida ({yml_path}):\n{exc}") from exc
    if not raw:
        raise ValueError(f"Config de clientes vazia ou inválida: {yml_path}")
    clients: dict[str, ClientConfig] = {}
    for name, data in raw.items():
        data["client"] = name
        try:
            clients[name] = ClientConfig.model_validate(data)
        except ValidationError as exc:
            campos = ", ".join(str(e["loc"][-1]) for e in exc.errors())
            raise ValueError(
                f"Cliente '{name}' com config inválida em clients.yml "
                f"(campos: {campos}):\n{exc}"
            ) from exc
    return clients

Core

tap_ixc.core.pipeline.PipelineRun

Executa stages em ordem, respeitando checkpoints.

Source code in tap_ixc/core/pipeline.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class PipelineRun:
    """
    Executa stages em ordem, respeitando checkpoints.
    """

    def __init__(
        self,
        client: str,
        system: str,
        pipeline: str,
        checkpoint: Checkpoint,
        events: EventStore,
        from_checkpoint: bool = False,
    ) -> None:
        self._client = client
        self._system = system
        self._pipeline = pipeline
        self._checkpoint = checkpoint
        self._events = events
        self._from_checkpoint = from_checkpoint
        self.ctx = PipelineContext()

    def _last_done_stage(self) -> Stage | None:
        cp = self._checkpoint.get_last(self._client, self._pipeline)
        if not cp:
            return None
        try:
            return Stage(cp["stage"])
        except ValueError:
            return None

    def _should_skip(self, stage: Stage, last_done: Stage | None) -> bool:
        if not self._from_checkpoint or last_done is None:
            return False
        last_idx = _STAGE_ORDER.index(last_done)
        this_idx = _STAGE_ORDER.index(stage)
        return this_idx <= last_idx

    def execute(self, stages: dict[Stage, StageFn]) -> PipelineContext:
        # Exposto como atributo p/ o caller ler contagens parciais mesmo se execute() falhar.
        self.ctx = ctx = PipelineContext()
        run_id = self._events.start_run(self._client, self._system, self._pipeline)
        ctx.set("run_id", run_id)
        last_done = self._last_done_stage() if self._from_checkpoint else None

        try:
            for stage in _STAGE_ORDER:
                if stage not in stages:
                    continue

                if self._should_skip(stage, last_done):
                    log.info("stage.skipped", stage=stage.value, pipeline=self._pipeline)
                    self._events.emit(run_id, "stage_skipped", stage=stage.value)
                    continue

                log.info("stage.start", stage=stage.value, pipeline=self._pipeline)
                self._events.emit(run_id, "stage_start", stage=stage.value)

                stages[stage](ctx)

                self._checkpoint.mark_done(
                    self._client,
                    self._pipeline,
                    stage.value,
                    data_path=ctx.get("data_path"),
                )
                self._events.emit(
                    run_id,
                    "stage_done",
                    stage=stage.value,
                    records=ctx.get("records_extracted"),
                )
                log.info("stage.done", stage=stage.value, pipeline=self._pipeline)

            self._events.finish_run(
                run_id,
                status="SUCCESS",
                records_in=ctx.get("records_extracted", 0),
                records_out=ctx.get("records_loaded", 0),
                stage=Stage.VERIFY.value,
            )
        except Exception as exc:
            safe_err = redact(str(exc))
            log.error("pipeline.failed", pipeline=self._pipeline, error=safe_err)
            self._events.finish_run(
                run_id,
                status="FAILED",
                error=safe_err,
                records_in=ctx.get("records_extracted", 0),
            )
            raise

        return ctx

tap_ixc.core.checkpoint.Checkpoint

Source code in tap_ixc/core/checkpoint.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class Checkpoint:
    def __init__(self, dsn: str, schema: str = "etl"):
        self._dsn = dsn
        self._schema = schema

    def mark_done(
        self,
        client: str,
        pipeline: str,
        stage: str,
        data_path: str | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        with psycopg.connect(self._dsn) as conn:
            conn.execute(
                f"""
                INSERT INTO {self._schema}.checkpoints (client, pipeline, stage, data_path, metadata)
                VALUES (%s, %s, %s, %s, %s)
                ON CONFLICT (client, pipeline)
                DO UPDATE SET
                    stage        = EXCLUDED.stage,
                    completed_at = now(),
                    data_path    = EXCLUDED.data_path,
                    metadata     = EXCLUDED.metadata
                """,
                (client, pipeline, stage, data_path, json.dumps(metadata or {})),
            )
        log.info("checkpoint.marked", client=client, pipeline=pipeline, stage=stage)

    def get_last(self, client: str, pipeline: str) -> dict[str, Any] | None:
        with psycopg.connect(self._dsn) as conn:
            row = conn.execute(
                f"""
                SELECT stage, data_path, metadata
                FROM {self._schema}.checkpoints
                WHERE client = %s AND pipeline = %s
                """,
                (client, pipeline),
            ).fetchone()
        if row:
            return {"stage": row[0], "data_path": row[1], "metadata": row[2]}
        return None

    def clear(self, client: str, pipeline: str) -> None:
        with psycopg.connect(self._dsn) as conn:
            conn.execute(
                f"DELETE FROM {self._schema}.checkpoints WHERE client = %s AND pipeline = %s",
                (client, pipeline),
            )
        log.info("checkpoint.cleared", client=client, pipeline=pipeline)

tap_ixc.core.events.EventStore

Source code in tap_ixc/core/events.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class EventStore:
    def __init__(self, dsn: str, schema: str = "etl"):
        self._dsn = dsn
        self._schema = schema

    def start_run(
        self,
        client: str,
        system: str,
        pipeline: str,
        metadata: dict[str, Any] | None = None,
    ) -> int:
        with psycopg.connect(self._dsn) as conn:
            row = conn.execute(
                f"""
                INSERT INTO {self._schema}.pipeline_runs (client, system, pipeline, status, metadata)
                VALUES (%s, %s, %s, 'RUNNING', %s)
                RETURNING id
                """,
                (client, system, pipeline, json.dumps(metadata or {})),
            ).fetchone()
        assert row is not None
        run_id = row[0]
        log.info("run.started", client=client, system=system, pipeline=pipeline, run_id=run_id)
        return run_id

    def finish_run(
        self,
        run_id: int,
        status: str = "SUCCESS",
        records_in: int = 0,
        records_out: int = 0,
        error: str | None = None,
        stage: str | None = None,
    ) -> None:
        with psycopg.connect(self._dsn) as conn:
            conn.execute(
                f"""
                UPDATE {self._schema}.pipeline_runs
                SET status      = %s,
                    finished_at = now(),
                    duration_s  = EXTRACT(EPOCH FROM (now() - started_at)),
                    records_in  = %s,
                    records_out = %s,
                    error       = %s,
                    stage       = %s
                WHERE id = %s
                """,
                (status, records_in, records_out, error, stage, run_id),
            )
        log.info("run.finished", run_id=run_id, status=status, records_out=records_out)

    def write_dead_letters(
        self,
        run_id: int,
        stream: str,
        dead: list[dict[str, Any]],
    ) -> None:
        """Grava linhas reprovadas na validação. dead = [{record, errors}, ...]."""
        if not dead:
            return
        with psycopg.connect(self._dsn) as conn:
            with conn.cursor() as cur:
                cur.executemany(
                    f"""
                    INSERT INTO {self._schema}.dead_letters (run_id, stream, record, errors)
                    VALUES (%s, %s, %s, %s)
                    """,
                    [
                        (
                            run_id,
                            stream,
                            json.dumps(d["record"], default=str),
                            json.dumps(d["errors"], default=str),
                        )
                        for d in dead
                    ],
                )
        log.warning("dead_letters.written", run_id=run_id, stream=stream, count=len(dead))

    def emit(
        self,
        run_id: int,
        event_type: str,
        stage: str | None = None,
        message: str | None = None,
        records: int | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        with psycopg.connect(self._dsn) as conn:
            conn.execute(
                f"""
                INSERT INTO {self._schema}.pipeline_events
                    (run_id, event_type, stage, message, records, metadata)
                VALUES (%s, %s, %s, %s, %s, %s)
                """,
                (run_id, event_type, stage, message, records, json.dumps(metadata or {})),
            )
        log.debug("event.emitted", run_id=run_id, event_type=event_type, stage=stage)

write_dead_letters(run_id, stream, dead)

Grava linhas reprovadas na validação. dead = [{record, errors}, ...].

Source code in tap_ixc/core/events.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def write_dead_letters(
    self,
    run_id: int,
    stream: str,
    dead: list[dict[str, Any]],
) -> None:
    """Grava linhas reprovadas na validação. dead = [{record, errors}, ...]."""
    if not dead:
        return
    with psycopg.connect(self._dsn) as conn:
        with conn.cursor() as cur:
            cur.executemany(
                f"""
                INSERT INTO {self._schema}.dead_letters (run_id, stream, record, errors)
                VALUES (%s, %s, %s, %s)
                """,
                [
                    (
                        run_id,
                        stream,
                        json.dumps(d["record"], default=str),
                        json.dumps(d["errors"], default=str),
                    )
                    for d in dead
                ],
            )
    log.warning("dead_letters.written", run_id=run_id, stream=stream, count=len(dead))

tap_ixc.core.contracts.sanitize(record)

Substitui strings vazias e datas inválidas (0000-00-00, 2024-03-00, etc.) por None. Migrado do sistema legado.

Source code in tap_ixc/core/contracts.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def sanitize(record: dict[str, Any]) -> dict[str, Any]:
    """
    Substitui strings vazias e datas inválidas (0000-00-00, 2024-03-00, etc.) por None.
    Migrado do sistema legado.
    """
    out: dict[str, Any] = {}
    for k, v in record.items():
        if isinstance(v, str):
            stripped = v.strip()
            if not stripped:
                out[k] = None
                continue
            if _DATE_PREFIX.match(stripped):
                parts = stripped[:10].split("-")
                if len(parts) == 3:
                    y, m, d = parts
                    if y == "0000" or m == "00" or d == "00":
                        out[k] = None
                        continue
                    try:
                        datetime.strptime(stripped[:10], "%Y-%m-%d")
                    except ValueError:
                        out[k] = None
                        continue
        out[k] = v
    return out

tap_ixc.core.contracts.validate_batch(records, schema=None)

Sanitiza e valida um batch. Retorna (válidos, dead_letter). Se schema=None, apenas sanitiza.

Source code in tap_ixc/core/contracts.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def validate_batch(
    records: list[dict[str, Any]],
    schema: Type[BaseModel] | None = None,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    """
    Sanitiza e valida um batch.
    Retorna (válidos, dead_letter).
    Se schema=None, apenas sanitiza.
    """
    valid: list[dict[str, Any]] = []
    dead: list[dict[str, Any]] = []

    for raw in records:
        clean = sanitize(raw)
        if schema is None:
            valid.append(clean)
            continue
        try:
            obj = schema.model_validate(clean)
            valid.append(obj.model_dump())
        except ValidationError as exc:
            dead.append({"record": clean, "errors": exc.errors()})

    if dead:
        log.warning("contracts.dead_letter", count=len(dead), total=len(records))

    return valid, dead

Loaders

tap_ixc.loaders.staging.StagingLoader

Consome um Iterator de records, grava NDJSON e carrega no DuckDB persistente.

Se fields for fornecido, o DuckDB só mantém essas colunas. Se transform_sql for fornecido, substitui qualquer SELECT automático.

Source code in tap_ixc/loaders/staging.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class StagingLoader:
    """
    Consome um Iterator de records, grava NDJSON e carrega no DuckDB persistente.

    Se `fields` for fornecido, o DuckDB só mantém essas colunas.
    Se `transform_sql` for fornecido, substitui qualquer SELECT automático.
    """

    def __init__(
        self,
        duckdb_path: str,
        table: str,
        fields: list[str] | None = None,
        transform_sql: str | None = None,
    ) -> None:
        self._duckdb_path = duckdb_path
        self._table = validate_identifier(table, "nome de tabela")
        self._fields = fields
        self._transform_sql = transform_sql
        os.makedirs(os.path.dirname(duckdb_path), exist_ok=True)

    def _ndjson_path(self) -> Path:
        base = Path(self._duckdb_path).parent / "ndjson"
        base.mkdir(parents=True, exist_ok=True)
        return base / f"{self._table}.ndjson"

    def _select_sql(self, raw_glob: str) -> str:
        source = f"read_json_auto('{raw_glob}')"
        if self._transform_sql:
            return self._transform_sql.replace("{raw}", source)
        if self._fields:
            cols = ", ".join(f'"{f}"' for f in self._fields)
            return f"SELECT {cols} FROM {source}"
        return f"SELECT * FROM {source}"

    def load(self, records: Iterator[dict[str, Any]]) -> tuple[str, int]:
        """
        Consome o iterador, grava NDJSON sanitizado e carrega no DuckDB.
        Retorna (ndjson_path, total_records).
        """
        ndjson_path = self._ndjson_path()
        total = 0

        with open(ndjson_path, "w", encoding="utf-8") as f:
            for rec in records:
                clean = sanitize(rec)
                f.write(json.dumps(clean, ensure_ascii=False))
                f.write("\n")
                total += 1

        if total == 0:
            log.warning("staging.empty", table=self._table)
            return str(ndjson_path), 0

        glob = str(ndjson_path).replace("\\", "/")
        conn = duckdb.connect(self._duckdb_path)
        try:
            select = self._select_sql(glob)
            conn.execute(f"CREATE OR REPLACE TABLE {self._table} AS ({select})")
        finally:
            conn.close()

        log.info("staging.loaded", table=self._table, records=total)
        return str(ndjson_path), total

    def read_all(self) -> list[dict[str, Any]]:
        """Lê todas as linhas da tabela DuckDB como dicts (para validação)."""
        conn = duckdb.connect(self._duckdb_path)
        try:
            cur = conn.execute(f"SELECT * FROM {self._table}")
            cols = [d[0] for d in cur.description]
            return [dict(zip(cols, row)) for row in cur.fetchall()]
        finally:
            conn.close()

    def replace(self, rows: list[dict[str, Any]]) -> None:
        """Reescreve a tabela DuckDB apenas com `rows` (subset pós-validação)."""
        ndjson_path = self._ndjson_path()
        with open(ndjson_path, "w", encoding="utf-8") as f:
            for r in rows:
                f.write(json.dumps(r, ensure_ascii=False, default=str))
                f.write("\n")

        conn = duckdb.connect(self._duckdb_path)
        try:
            if not rows:
                # todas reprovadas: zera mantendo o schema da tabela
                conn.execute(f"DELETE FROM {self._table}")
            else:
                glob = str(ndjson_path).replace("\\", "/")
                conn.execute(
                    f"CREATE OR REPLACE TABLE {self._table} AS "
                    f"(SELECT * FROM read_json_auto('{glob}'))"
                )
        finally:
            conn.close()

        log.info("staging.replaced", table=self._table, records=len(rows))

load(records)

Consome o iterador, grava NDJSON sanitizado e carrega no DuckDB. Retorna (ndjson_path, total_records).

Source code in tap_ixc/loaders/staging.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def load(self, records: Iterator[dict[str, Any]]) -> tuple[str, int]:
    """
    Consome o iterador, grava NDJSON sanitizado e carrega no DuckDB.
    Retorna (ndjson_path, total_records).
    """
    ndjson_path = self._ndjson_path()
    total = 0

    with open(ndjson_path, "w", encoding="utf-8") as f:
        for rec in records:
            clean = sanitize(rec)
            f.write(json.dumps(clean, ensure_ascii=False))
            f.write("\n")
            total += 1

    if total == 0:
        log.warning("staging.empty", table=self._table)
        return str(ndjson_path), 0

    glob = str(ndjson_path).replace("\\", "/")
    conn = duckdb.connect(self._duckdb_path)
    try:
        select = self._select_sql(glob)
        conn.execute(f"CREATE OR REPLACE TABLE {self._table} AS ({select})")
    finally:
        conn.close()

    log.info("staging.loaded", table=self._table, records=total)
    return str(ndjson_path), total

read_all()

Lê todas as linhas da tabela DuckDB como dicts (para validação).

Source code in tap_ixc/loaders/staging.py 
83
84
85
86
87
88
89
90
91
def read_all(self) -> list[dict[str, Any]]:
    """Lê todas as linhas da tabela DuckDB como dicts (para validação)."""
    conn = duckdb.connect(self._duckdb_path)
    try:
        cur = conn.execute(f"SELECT * FROM {self._table}")
        cols = [d[0] for d in cur.description]
        return [dict(zip(cols, row)) for row in cur.fetchall()]
    finally:
        conn.close()

replace(rows)

Reescreve a tabela DuckDB apenas com rows (subset pós-validação).

Source code in tap_ixc/loaders/staging.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def replace(self, rows: list[dict[str, Any]]) -> None:
    """Reescreve a tabela DuckDB apenas com `rows` (subset pós-validação)."""
    ndjson_path = self._ndjson_path()
    with open(ndjson_path, "w", encoding="utf-8") as f:
        for r in rows:
            f.write(json.dumps(r, ensure_ascii=False, default=str))
            f.write("\n")

    conn = duckdb.connect(self._duckdb_path)
    try:
        if not rows:
            # todas reprovadas: zera mantendo o schema da tabela
            conn.execute(f"DELETE FROM {self._table}")
        else:
            glob = str(ndjson_path).replace("\\", "/")
            conn.execute(
                f"CREATE OR REPLACE TABLE {self._table} AS "
                f"(SELECT * FROM read_json_auto('{glob}'))"
            )
    finally:
        conn.close()

    log.info("staging.replaced", table=self._table, records=len(rows))

tap_ixc.loaders.postgres.PostgresLoader

Source code in tap_ixc/loaders/postgres.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
class PostgresLoader:
    def __init__(
        self,
        duckdb_path: str,
        pg_dsn: str,
        schema: str,
        table: str,
        strategy: str = "full",
        pk_column: str = "id",
    ) -> None:
        self._duckdb_path = duckdb_path
        self._pg_dsn = pg_dsn
        self._schema = validate_identifier(schema, "schema")
        self._table = validate_identifier(table, "nome de tabela")
        self._strategy = strategy
        self._pk_column = validate_identifier(pk_column, "pk_column")

    @property
    def _stg(self) -> str:
        return f'pg."{self._schema}"."__stg_{self._table}"'

    @property
    def _qualified(self) -> str:
        return f'pg."{self._schema}"."{self._table}"'

    def drop_staging(self) -> None:
        """Remove a staging `__stg_<table>` do Postgres (idempotente).

        Para teardown: garante limpeza mesmo se o LOAD falhar antes do swap.
        Só fala com o Postgres → roda em qualquer worker.
        """
        conn = duckdb.connect()
        try:
            conn.execute("INSTALL postgres; LOAD postgres;")
            conn.execute(
                f"ATTACH '{self._pg_dsn}' AS pg (TYPE postgres, SCHEMA '{self._schema}')"
            )
            conn.execute(f"DROP TABLE IF EXISTS {self._stg}")
            conn.execute("DETACH pg")
        finally:
            conn.close()

    def stage(self) -> int:
        """Empurra a tabela DuckDB local para a staging compartilhada no Postgres
        (`__stg_<table>`). Roda no worker que tem o DuckDB (o do EXTRACT). Retorna count.
        """
        conn = duckdb.connect(self._duckdb_path)
        try:
            conn.execute("INSTALL postgres; LOAD postgres;")
            conn.execute("SET pg_null_byte_replacement='';")
            conn.execute(
                f"ATTACH '{self._pg_dsn}' AS pg (TYPE postgres, SCHEMA '{self._schema}')"
            )
            conn.execute(f"CREATE OR REPLACE TABLE {self._stg} AS SELECT * FROM {self._table}")
            count: int = conn.execute(f"SELECT count(*) FROM {self._stg}").fetchone()[0]  # type: ignore[index]
            conn.execute("DETACH pg")
        finally:
            conn.close()
        log.info("postgres.staged", table=self._table, schema=self._schema, records=count)
        return count

    def swap(self) -> int:
        """Troca a staging do Postgres (`__stg_<table>`) para a tabela final, atômico.
        Só fala com o Postgres (DuckDB em memória) → roda em QUALQUER worker. Retorna count.
        """
        conn = duckdb.connect()
        try:
            conn.execute("INSTALL postgres; LOAD postgres;")
            conn.execute(
                f"ATTACH '{self._pg_dsn}' AS pg (TYPE postgres, SCHEMA '{self._schema}')"
            )
            qualified, stg_remote = self._qualified, self._stg
            count: int = conn.execute(f"SELECT count(*) FROM {stg_remote}").fetchone()[0]  # type: ignore[index]

            try:
                conn.execute(f"SELECT 1 FROM {qualified} LIMIT 0")
                table_exists = True
            except duckdb.Error:
                table_exists = False

            try:
                conn.execute("BEGIN;")
                if self._strategy == "full" or not table_exists:
                    conn.execute(f"DROP TABLE IF EXISTS {qualified}")
                    conn.execute(f"CREATE TABLE {qualified} AS SELECT * FROM {stg_remote}")
                else:  # delta — evolui schema (da própria staging) e insere por nome
                    target_cols = {
                        d[0] for d in conn.execute(f"SELECT * FROM {qualified} LIMIT 0").description
                    }
                    stg_schema = {
                        r[0]: r[1] for r in conn.execute(f"DESCRIBE {stg_remote}").fetchall()
                    }
                    for col in _column_plan(target_cols, list(stg_schema)):
                        pgtype = _pg_type(stg_schema[col])
                        conn.execute(f'ALTER TABLE {qualified} ADD COLUMN IF NOT EXISTS "{col}" {pgtype}')
                        log.warning("postgres.schema_evolved", table=self._table, column=col, type=pgtype)
                    cols = ", ".join(f'"{c}"' for c in stg_schema)
                    conn.execute(
                        f"""DELETE FROM {qualified} AS tgt USING {stg_remote} AS src
                            WHERE tgt."{self._pk_column}" = src."{self._pk_column}";"""
                    )
                    conn.execute(f"INSERT INTO {qualified} ({cols}) SELECT {cols} FROM {stg_remote}")
                conn.execute("COMMIT;")
            except Exception:
                try:
                    conn.execute("ROLLBACK;")
                except Exception:
                    pass
                raise
            finally:
                try:
                    conn.execute(f"DROP TABLE IF EXISTS {stg_remote}")
                except Exception:
                    pass
                conn.execute("DETACH pg")
        finally:
            conn.close()
        log.info(
            "postgres.loaded", table=self._table, schema=self._schema,
            strategy=self._strategy, records=count,
        )
        return count

    def load(self) -> int:
        """stage + swap num passo só (usado por runner.run / cron — staging local)."""
        self.stage()
        return self.swap()

drop_staging()

Remove a staging __stg_<table> do Postgres (idempotente).

Para teardown: garante limpeza mesmo se o LOAD falhar antes do swap. Só fala com o Postgres → roda em qualquer worker.

Source code in tap_ixc/loaders/postgres.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def drop_staging(self) -> None:
    """Remove a staging `__stg_<table>` do Postgres (idempotente).

    Para teardown: garante limpeza mesmo se o LOAD falhar antes do swap.
    Só fala com o Postgres → roda em qualquer worker.
    """
    conn = duckdb.connect()
    try:
        conn.execute("INSTALL postgres; LOAD postgres;")
        conn.execute(
            f"ATTACH '{self._pg_dsn}' AS pg (TYPE postgres, SCHEMA '{self._schema}')"
        )
        conn.execute(f"DROP TABLE IF EXISTS {self._stg}")
        conn.execute("DETACH pg")
    finally:
        conn.close()

stage()

Empurra a tabela DuckDB local para a staging compartilhada no Postgres (__stg_<table>). Roda no worker que tem o DuckDB (o do EXTRACT). Retorna count.

Source code in tap_ixc/loaders/postgres.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def stage(self) -> int:
    """Empurra a tabela DuckDB local para a staging compartilhada no Postgres
    (`__stg_<table>`). Roda no worker que tem o DuckDB (o do EXTRACT). Retorna count.
    """
    conn = duckdb.connect(self._duckdb_path)
    try:
        conn.execute("INSTALL postgres; LOAD postgres;")
        conn.execute("SET pg_null_byte_replacement='';")
        conn.execute(
            f"ATTACH '{self._pg_dsn}' AS pg (TYPE postgres, SCHEMA '{self._schema}')"
        )
        conn.execute(f"CREATE OR REPLACE TABLE {self._stg} AS SELECT * FROM {self._table}")
        count: int = conn.execute(f"SELECT count(*) FROM {self._stg}").fetchone()[0]  # type: ignore[index]
        conn.execute("DETACH pg")
    finally:
        conn.close()
    log.info("postgres.staged", table=self._table, schema=self._schema, records=count)
    return count

swap()

Troca a staging do Postgres (__stg_<table>) para a tabela final, atômico. Só fala com o Postgres (DuckDB em memória) → roda em QUALQUER worker. Retorna count.

Source code in tap_ixc/loaders/postgres.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def swap(self) -> int:
    """Troca a staging do Postgres (`__stg_<table>`) para a tabela final, atômico.
    Só fala com o Postgres (DuckDB em memória) → roda em QUALQUER worker. Retorna count.
    """
    conn = duckdb.connect()
    try:
        conn.execute("INSTALL postgres; LOAD postgres;")
        conn.execute(
            f"ATTACH '{self._pg_dsn}' AS pg (TYPE postgres, SCHEMA '{self._schema}')"
        )
        qualified, stg_remote = self._qualified, self._stg
        count: int = conn.execute(f"SELECT count(*) FROM {stg_remote}").fetchone()[0]  # type: ignore[index]

        try:
            conn.execute(f"SELECT 1 FROM {qualified} LIMIT 0")
            table_exists = True
        except duckdb.Error:
            table_exists = False

        try:
            conn.execute("BEGIN;")
            if self._strategy == "full" or not table_exists:
                conn.execute(f"DROP TABLE IF EXISTS {qualified}")
                conn.execute(f"CREATE TABLE {qualified} AS SELECT * FROM {stg_remote}")
            else:  # delta — evolui schema (da própria staging) e insere por nome
                target_cols = {
                    d[0] for d in conn.execute(f"SELECT * FROM {qualified} LIMIT 0").description
                }
                stg_schema = {
                    r[0]: r[1] for r in conn.execute(f"DESCRIBE {stg_remote}").fetchall()
                }
                for col in _column_plan(target_cols, list(stg_schema)):
                    pgtype = _pg_type(stg_schema[col])
                    conn.execute(f'ALTER TABLE {qualified} ADD COLUMN IF NOT EXISTS "{col}" {pgtype}')
                    log.warning("postgres.schema_evolved", table=self._table, column=col, type=pgtype)
                cols = ", ".join(f'"{c}"' for c in stg_schema)
                conn.execute(
                    f"""DELETE FROM {qualified} AS tgt USING {stg_remote} AS src
                        WHERE tgt."{self._pk_column}" = src."{self._pk_column}";"""
                )
                conn.execute(f"INSERT INTO {qualified} ({cols}) SELECT {cols} FROM {stg_remote}")
            conn.execute("COMMIT;")
        except Exception:
            try:
                conn.execute("ROLLBACK;")
            except Exception:
                pass
            raise
        finally:
            try:
                conn.execute(f"DROP TABLE IF EXISTS {stg_remote}")
            except Exception:
                pass
            conn.execute("DETACH pg")
    finally:
        conn.close()
    log.info(
        "postgres.loaded", table=self._table, schema=self._schema,
        strategy=self._strategy, records=count,
    )
    return count

load()

stage + swap num passo só (usado por runner.run / cron — staging local).

Source code in tap_ixc/loaders/postgres.py 
166
167
168
169
def load(self) -> int:
    """stage + swap num passo só (usado por runner.run / cron — staging local)."""
    self.stage()
    return self.swap()

Runner

tap_ixc.runner.run(client_name, streams=None, from_checkpoint=False, duckdb_path=None)

Roda o tap para um cliente definido no clients.yml. streams=None → todos os streams configurados no YAML. duckdb_path → sobrescreve o staging DuckDB (útil p/ isolar por stream em execuções paralelas, ex: uma task Airflow por stream).

Source code in tap_ixc/runner.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def run(
    client_name: str,
    streams: list[str] | None = None,
    from_checkpoint: bool = False,
    duckdb_path: str | None = None,
) -> list[TapResult]:
    """
    Roda o tap para um cliente definido no clients.yml.
    streams=None → todos os streams configurados no YAML.
    duckdb_path → sobrescreve o staging DuckDB (útil p/ isolar por stream em
        execuções paralelas, ex: uma task Airflow por stream).
    """
    cfg = get_client(client_name)
    tap = IXCTap(cfg.api)
    destination = Destination(
        postgres_dsn=cfg.postgres_dsn,
        schema=cfg.schema_name,
        duckdb_path=duckdb_path or cfg.duckdb_resolved(),
    )

    catalog = _build_catalog_from_config(cfg)

    if streams:
        catalog = catalog.select(*streams)

    results = tap.sync(destination, catalog, from_checkpoint, client_name)

    if cfg.webhook_url:
        _notify_webhook(cfg.webhook_url, client_name, results)

    return results