Metadata-Version: 2.4
Name: ducklake-cdc-client
Version: 0.6.1
Summary: Python client helpers for the ducklake-cdc DuckDB extension
Requires-Python: >=3.14
Requires-Dist: duckdb==1.5.4
Requires-Dist: ducklake-client>=0.8.1
Requires-Dist: pytz>=2026.2
Provides-Extra: arrow
Requires-Dist: pyarrow; extra == 'arrow'
Provides-Extra: pandas
Requires-Dist: pandas; extra == 'pandas'
Provides-Extra: polars
Requires-Dist: polars; extra == 'polars'
Requires-Dist: pyarrow; extra == 'polars'
Description-Content-Type: text/markdown

# ducklake-cdc-client

Python client helpers for the `ducklake_cdc` DuckDB community extension.

The package gives you two layers:

- `CDCClient`: a direct Python wrapper over the extension's SQL table functions.
- `DMLConsumer` / `DDLConsumer`: durable consumers that yield batches and commit only after
  your code has processed them.

## Install

```sh
pip install ducklake-cdc-client
```

The package uses [`ducklake-client`](https://pypi.org/project/ducklake-client/) for DuckLake
connections. `CDCClient` installs and loads the DuckDB community extension on first use:

```sql
INSTALL ducklake_cdc FROM community;
LOAD ducklake_cdc;
```

## Batch iteration

```python
from ducklake_client import DiskStorage, DuckDBCatalog, DuckLake
from ducklake_cdc_client import DMLConsumer

with DuckLake(
    catalog=DuckDBCatalog("metadata.ducklake"),
    storage=DiskStorage("data"),
) as lake:
    with DMLConsumer(
        lake,
        "orders-consumer",
        table="main.orders",
        mode="changes",
    ) as consumer:
        for batch in consumer.batches(infinite=False):
            for change in batch:
                print(change.to_dict())
            batch.commit()
```

`batch.commit()` advances the durable consumer cursor. If processing raises before that call,
the same batch can be read again on the next run.

## Connections, retries, and restart safety

Each high-level consumer uses one dedicated DuckDB connection for create/read/listen,
heartbeats, and commit. This is required because the extension's lease belongs to the
connection that acquired it. By default the client derives and owns that connection. If you
pass `connection=` or `client=`, dedicate its connection to that one consumer: do not run
unrelated queries on it, because cancellation calls `connection.interrupt()`.

Known SQLite lock bursts, every observed H-022 deadlock spelling (`thread::join failed`,
`resource deadlock would occur`, and `resource deadlock avoided`), and typed lease contention
errors use the default bounded retry policy. Retry is not a recovery boundary for a poisoned
DuckDB handle. After retries are exhausted, discard the consumer and its connection, open a new
consumer instance, and resume the same durable consumer name. Call `prewarm()` on every handle
immediately after loading the extension and before other catalog activity.

Lease failures are catchable as `LeaseContentionError` or `LeaseTimeoutError`; both inherit
`RetryableCDCError`. A process supervisor should back off and reopen on a fresh dedicated
connection. Reopening with `on_exists="use"` resumes from the last committed snapshot, so only
commit after sink processing succeeds.

Graceful shutdown is bounded:

```python
consumer.close(timeout=5.0, cancel=True, release=True)
```

`cancel=True` interrupts an active listen/read and that caller receives
`ConsumerCancelledError`. `cancel=False` waits without interrupting. Either mode raises
`ConsumerCloseTimeoutError` if the deadline expires. `release=True` calls the owner-token-
conditional `cdc_consumer_release`, so a stale close cannot clear a successor connection's
lease. Use `release=False` only when an external supervisor owns lease cleanup. The
unconditional `cdc_consumer_force_release` remains an operator recovery command. Abrupt process
death cannot run `close()`: the next process must wait for lease expiry or use
`lease_policy="takeover"` only after it knows the previous holder is dead.

`cdc_consumer_release` requires `ducklake_cdc >= 0.5.4`. With an older extension, high-level
close safely closes its owned connection and relies on lease expiry; it never falls back to
unconditional force release. Upgrade the extension to make graceful release immediate.

Cancellation ends the current run and is never retried in place. A lease timeout is marked
retryable for supervisors but likewise does not repeat its entire wait internally; reopen after
backoff instead.

## Sink-driven usage

If you prefer a push style, pass sinks and let `consumer.run()` deliver and commit for you.

```python
from ducklake_client import DiskStorage, DuckDBCatalog, DuckLake
from ducklake_cdc_client import DMLConsumer, StdoutSink

with DuckLake(
    catalog=DuckDBCatalog("metadata.ducklake"),
    storage=DiskStorage("data"),
) as lake:
    with DMLConsumer(
        lake,
        "orders-consumer",
        table="main.orders",
        mode="changes",
        sinks=[StdoutSink()],
    ) as consumer:
        consumer.run(infinite=False)
```

## Demo

Run the local demo:

```sh
uv run python demo.py
```

The demo creates a local DuckLake catalog under `.demo/`, inserts one row into `main.orders`,
prints the emitted CDC change batch, and commits it.

## Test with a local extension build

The extension binary must match the Python package's DuckDB version exactly. This checkout pins
DuckDB `1.5.4`. After building the sibling extension repo, copy its unsigned macOS/arm64 binary
into this repo's ignored local-artifact directory:

```sh
mkdir -p .local/extensions/v1.5.4/osx_arm64
cp ../ducklake-cdc-extension/build/release/extension/ducklake_cdc/ducklake_cdc.duckdb_extension \
  .local/extensions/v1.5.4/osx_arm64/
export DUCKLAKE_CDC_EXTENSION="$PWD/.local/extensions/v1.5.4/osx_arm64/ducklake_cdc.duckdb_extension"
```

Allow unsigned extensions when DuckDB opens, load the binary before constructing the consumer,
and let the client skip its normal community-repository install:

```python
import os

from ducklake_client import DiskStorage, DuckDBCatalog, DuckDBConfig, DuckLake
from ducklake_cdc_client import CDCClient, DMLConsumer

with DuckLake(
    catalog=DuckDBCatalog("metadata.ducklake"),
    storage=DiskStorage("data"),
    duckdb=DuckDBConfig(config={"allow_unsigned_extensions": True}),
) as lake:
    lake.connection.execute(f"LOAD '{os.environ['DUCKLAKE_CDC_EXTENSION']}'")
    client = CDCClient(lake, install_extension=False)
    with DMLConsumer(
        lake,
        "orders-consumer",
        table="main.orders",
        mode="changes",
        client=client,
    ) as consumer:
        print(consumer.client.version())
```

This binary is platform-specific (`osx_arm64`) and unsigned; rebuild it for another DuckDB
version or platform instead of reusing it.
