Metadata-Version: 2.4
Name: databricks-zerobus-ingest-sdk
Version: 1.11.0
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: Intended Audience :: System Administrators
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Operating System :: OS Independent
Requires-Dist: protobuf>=4.25.0,<7.0
Requires-Dist: requests>=2.28.1,<3
Requires-Dist: certifi>=2026.10.19
Requires-Dist: pyarrow>=14.0.0,<22.0 ; python_full_version < '3.14' and extra == 'arrow'
Requires-Dist: pyarrow>=22.0.0,<26.0 ; python_full_version >= '3.14' and extra == 'arrow'
Requires-Dist: fastavro>=1.9,<2 ; extra == 'avro'
Requires-Dist: wheel ; extra == 'dev'
Requires-Dist: build ; extra == 'dev'
Requires-Dist: grpcio-tools>=1.60.0,<2.0 ; extra == 'dev'
Requires-Dist: pip-tools>=7.0,<8 ; extra == 'dev'
Requires-Dist: black ; extra == 'dev'
Requires-Dist: pycodestyle ; extra == 'dev'
Requires-Dist: autoflake ; extra == 'dev'
Requires-Dist: isort ; extra == 'dev'
Requires-Dist: pytest ; extra == 'dev'
Requires-Dist: pytest-cov ; extra == 'dev'
Requires-Dist: pytest-xdist>=3.6.1,<4.0 ; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21,<1.0 ; extra == 'dev'
Requires-Dist: pyarrow>=14.0.0,<22.0 ; python_full_version < '3.14' and extra == 'dev'
Requires-Dist: pyarrow>=22.0.0,<26.0 ; python_full_version >= '3.14' and extra == 'dev'
Requires-Dist: fastavro>=1.9,<2 ; extra == 'dev'
Provides-Extra: arrow
Provides-Extra: avro
Provides-Extra: dev
License-File: LICENSE
License-File: NOTICE
Summary: Databricks Zerobus Ingest SDK for Python
Keywords: zerobus,databricks,sdk
Requires-Python: >=3.9, <3.15
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM

# Zerobus Python SDK

[![PyPI - Downloads](https://img.shields.io/pypi/dw/databricks-zerobus-ingest-sdk)](https://pypistats.org/packages/databricks-zerobus-ingest-sdk)
[![PyPI - License](https://img.shields.io/pypi/l/databricks-zerobus-ingest-sdk)](https://github.com/databricks/zerobus-sdk/blob/main/LICENSE)
![PyPI](https://img.shields.io/pypi/v/databricks-zerobus-ingest-sdk)

A high-performance Python client for streaming data ingestion into Databricks Delta tables using the Zerobus service.

## Table of Contents

- [Overview](#overview)
- [Features](#features)
- [Installation](#installation)
- [Quick Start](#quick-start)
  - [JSON (Simplest)](#option-1-json-simplest)
  - [Protocol Buffers](#option-2-protocol-buffers)
  - [Acknowledgments and throughput](#acknowledgments-and-throughput)
- [Persistent Streams](#persistent-streams)
- [Configuration](#configuration)
- [Error Handling](#error-handling)
- [Handling Stream Failures](#handling-stream-failures)
- [Performance Tips](#performance-tips)
- [API Reference](#api-reference)
- [Debugging](#debugging)
- [Building from Source](#building-from-source)
- [Community and Contributing](#community-and-contributing)
- [License](#license)

## Overview

The Zerobus Python SDK is a thin wrapper around the [Zerobus Rust SDK](../rust/), built using PyO3 bindings. It delivers native performance with a Python-friendly API supporting both synchronous and asynchronous usage.

**What is Zerobus?** See the [project overview](https://github.com/databricks/zerobus-sdk/blob/main/README.md#what-is-zerobus) for details on the Zerobus service.

**Prerequisites** (workspace setup, table creation, service principal): See the [top-level README](https://github.com/databricks/zerobus-sdk/blob/main/README.md#prerequisites).

### Architecture

```
┌─────────────────────────────────────────┐
│         Python Application Code         │
└─────────────────────────────────────────┘
                    │
                    ▼
┌─────────────────────────────────────────┐
│       Python SDK (Thin Wrapper)         │
│    • Sync and async APIs                │
│    • Python types & error handling      │
└─────────────────────────────────────────┘
                    │
                    ▼ (PyO3 bindings)
┌─────────────────────────────────────────┐
│         Rust Core Implementation        │
│    • gRPC communication                 │
│    • OAuth 2.0 authentication           │
│    • Stream management & recovery       │
└─────────────────────────────────────────┘
```

## Features

- **Rust-backed performance** - Native Rust implementation via PyO3 bindings for maximum throughput
- **Sync and Async support** - Both synchronous and asynchronous Python APIs
- **Automatic recovery** - Built-in retry and reconnection for transient failures
- **Multiple serialization formats** - JSON, Protocol Buffers, Avro, and Arrow Flight ingestion
- **OAuth 2.0 authentication** - Secure authentication with client credentials, automatically refreshed
- **Acknowledgment callbacks** - Receive notifications when records are acknowledged or encounter errors
- **Flexible configuration** - Fine-tune timeouts, retries, and recovery behavior

## Installation

### From PyPI (Recommended)

```bash
pip install databricks-zerobus-ingest-sdk
```

Pre-built wheels are available for:

- **Linux**: x86_64, aarch64 (manylinux)
- **macOS**: x86_64, arm64
- **Windows**: x86_64

### Python Version

Requires **Python 3.9-3.14**. The SDK is tested on CPython 3.9 through 3.14.

The wheel uses the CPython stable ABI (`abi3`), so one wheel works on every
supported version. The free-threaded builds (like `3.14t`) are not
supported, because the stable ABI does not cover them.

### Dependencies

- `protobuf` >= 4.25.0, < 7.0 (for Protocol Buffer schema handling)
- `requests` >= 2.28.1, < 3 (only for the `generate_proto` utility tool)

All core ingestion functionality (gRPC, OAuth, stream management) is handled by the native Rust implementation.

Arrow Flight ingestion needs `pyarrow`. Install it with the `arrow` extra:

```bash
pip install "databricks-zerobus-ingest-sdk[arrow]"
```

The extra selects the `pyarrow` version for your interpreter, because no single
`pyarrow` release covers Python 3.9 through 3.14. On Python 3.14 the extra
installs `pyarrow` 22.0.0 or later, which is the first release with 3.14 wheels.
On earlier versions it installs `pyarrow` below 22.0.0. Core ingestion (Protobuf
and JSON) does not need `pyarrow` at all.

Avro record format is built in (beta). Encoding **dict** records needs `fastavro`, installed
with the `avro` extra; pre-encoded Avro bytes need no extra:

```bash
pip install "databricks-zerobus-ingest-sdk[avro]"
```

## Quick Start

### Choose Your Serialization Format

1. **Protocol Buffers** (Recommended) - Strongly-typed schemas with compact binary encoding. More efficient over the wire and the best choice for production and high-throughput workloads.
2. **JSON** - Simple, no schema compilation needed. Good for getting started or quick prototyping, but each record carries higher per-record overhead (text serialization plus UTF-8 validation), so it is slower than Protocol Buffers for high-volume ingestion.
3. **Avro** (beta) - Row-oriented binary encoding against a declared writer schema. See the [sync](examples/sync_example_avro.py) or [async](examples/async_example_avro.py) example.
   - If you're sending **dict records**, they're encoded with `fastavro` — install the `avro` extra.
   - If you're sending **pre-encoded Avro bytes**, they pass through unchanged — no extra needed.
4. Arrow Flight - High-throughput columnar ingestion for applications that already produce Arrow data. Install the `arrow` extra and see the [sync](examples/sync_example_arrow.py) or [async](examples/async_example_arrow.py) example.

### Option 1: JSON (Simplest)

**Synchronous:**

```python
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import TableProperties

server_endpoint = "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com"
workspace_url = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com"

sdk = ZerobusSdk(server_endpoint, workspace_url)
table_properties = TableProperties("main.default.air_quality")
stream = sdk.create_stream(client_id, client_secret, table_properties)

try:
    for i in range(100):
        offset = stream.ingest_record_offset({
            "device_name": f"sensor-{i % 10}",
            "temp": 20 + (i % 15),
            "humidity": 50 + (i % 40)
        })
    stream.flush()
finally:
    stream.close()
```

**Asynchronous:**

```python
import asyncio
from zerobus.sdk.aio import ZerobusSdk
from zerobus.sdk.shared import TableProperties

async def main():
    server_endpoint = "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com"
    workspace_url = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com"

    sdk = ZerobusSdk(server_endpoint, workspace_url)
    table_properties = TableProperties("main.default.air_quality")
    stream = await sdk.create_stream(client_id, client_secret, table_properties)

    try:
        for i in range(100):
            offset = await stream.ingest_record_offset({
                "device_name": f"sensor-{i % 10}",
                "temp": 20 + (i % 15),
                "humidity": 50 + (i % 40)
            })
        await stream.flush()
    finally:
        await stream.close()

asyncio.run(main())
```

### Acknowledgments and throughput

Ingestion is asynchronous. `ingest_record_offset()` returns as soon as the record is
queued; the SDK sends it and tracks its acknowledgment in the background. To confirm
records are durably committed, call `flush()` — it returns once everything queued so far
is acknowledged. The idiomatic flow is **ingest in a loop, then `flush()`** (once for a
bounded batch, or periodically for a long-running stream); or register an
[`AckCallback`](#ackcallback) to be notified as records commit.

Each ingest also returns the record's offset, and `wait_for_offset(offset)` blocks until
that offset is acknowledged — handy when a specific record must be confirmed before
continuing (acks are ordered, so waiting on the last offset confirms the whole run). Just
avoid calling `wait_for_offset()` after every record in a tight loop, since that limits
throughput to one record per round-trip.

### Option 2: Protocol Buffers

First, define a protobuf schema. Use `proto2` syntax with `optional` fields to match Delta table columns:

```protobuf
// record.proto
syntax = "proto2";
message AirQuality {
    optional string device_name = 1;
    optional int32 temp = 2;
    optional int64 humidity = 3;
}
```

See the [Delta → Protobuf type mappings](https://github.com/databricks/zerobus-sdk/blob/main/README.md#delta--protobuf-type-mappings) in the top-level README.

**Compile the schema** to generate a Python module:

```bash
pip install "grpcio-tools>=1.60.0,<2.0"
python -m grpc_tools.protoc --python_out=. --proto_path=. record.proto
# Generates record_pb2.py
```

**Load the descriptor** from the generated module and pass it to `TableProperties`:

```python
import record_pb2

# The DESCRIPTOR is the compiled schema — pass it so the SDK can validate records
table_properties = TableProperties("main.default.air_quality", record_pb2.AirQuality.DESCRIPTOR)
```

Alternatively, generate the schema automatically from an existing Unity Catalog table:

```bash
python -m zerobus.tools.generate_proto \
    --uc-endpoint "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com" \
    --client-id "your-client-id" \
    --client-secret "your-client-secret" \
    --table "main.default.air_quality" \
    --output "record.proto" \
    --proto-msg "AirQuality"

# Then compile the generated file the same way:
python -m grpc_tools.protoc --python_out=. --proto_path=. record.proto
```

**Synchronous:**

```python
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import TableProperties
import record_pb2

sdk = ZerobusSdk(server_endpoint, workspace_url)
table_properties = TableProperties("main.default.air_quality", record_pb2.AirQuality.DESCRIPTOR)
stream = sdk.create_stream(client_id, client_secret, table_properties)

try:
    for i in range(100):
        record = record_pb2.AirQuality(
            device_name=f"sensor-{i % 10}",
            temp=20 + (i % 15),
            humidity=50 + (i % 40)
        )
        stream.ingest_record_offset(record)
    stream.flush()
finally:
    stream.close()
```

**Asynchronous:**

```python
import asyncio
from zerobus.sdk.aio import ZerobusSdk
from zerobus.sdk.shared import TableProperties
import record_pb2

async def main():
    sdk = ZerobusSdk(server_endpoint, workspace_url)
    table_properties = TableProperties("main.default.air_quality", record_pb2.AirQuality.DESCRIPTOR)
    stream = await sdk.create_stream(client_id, client_secret, table_properties)

    try:
        for i in range(100):
            record = record_pb2.AirQuality(
                device_name=f"sensor-{i % 10}",
                temp=20 + (i % 15),
                humidity=50 + (i % 40)
            )
            await stream.ingest_record_offset(record)
        await stream.flush()
    finally:
        await stream.close()

asyncio.run(main())
```

See the [`examples/`](examples/) directory for complete runnable examples.

## Authentication

`create_stream()` supports three authentication methods: `auth` (external-IdP
federation), a custom `headers_provider`, and `client_id`/`client_secret` (OAuth).
`auth` and `client_id`/`client_secret` are mutually exclusive — passing both
raises `ValueError` rather than silently preferring one. When `auth` is given
alongside `headers_provider`, `auth` takes precedence.

### OAuth client credentials (default)

```python
stream = sdk.create_stream(
    client_id="your-client-id",
    client_secret="your-client-secret",
    table_properties=table_properties,
)
```

### External-IdP federation (e.g. Entra ID)

Use `auth=FederatedToken(...)` to authenticate with an external identity
provider instead of a Databricks OAuth secret. You provide a callback that
returns the current external IdP token; the SDK exchanges it for a
Zerobus-scoped Databricks token (RFC 8693 token exchange) and caches and
refreshes that token for you. The callback may be synchronous (sync SDK) or
asynchronous (async SDK).

```python
from zerobus import FederatedToken

def get_idp_token():
    # Return the current external IdP (e.g. Entra ID) access token.
    ...

# Account-level federation: no Databricks service principal. The identity is
# synced into Databricks via Automatic Identity Management (SCIM). Build the
# FederatedToken once and reuse it across create_stream calls: the exchanged
# Databricks token is cached per token instance, so reusing one shares that cache
# (a distinct instance is a distinct identity and re-mints its own token).
auth = FederatedToken(idp_token_supplier=get_idp_token)
stream = sdk.create_stream(table_properties=table_properties, auth=auth)
another_stream = sdk.create_stream(table_properties=other_table, auth=auth)

# Workload identity federation: a Databricks service principal with a client_id
# and no secret, with a federation policy attached.
workload_auth = FederatedToken(
    idp_token_supplier=get_idp_token, databricks_client_id="<sp-client-id>"
)
stream = sdk.create_stream(table_properties=table_properties, auth=workload_auth)
```

The callback is invoked only when a fresh token must be minted (a cache miss or
refresh), not on every request, so keep it reasonably fast — ideally returning a
cached IdP token rather than doing a blocking network call on each invocation.

> **Note:** A **synchronous** callback runs inline while it blocks, holding the
> interpreter (GIL), so a slow blocking one stalls the SDK for the whole IdP
> round-trip and cannot be interrupted by `recovery_timeout_ms`. This holds on the
> **async** SDK too: a sync callback is invoked before the await point, so it
> blocks the event loop / a worker just as it blocks the sync SDK. For IdP calls
> that may do real network I/O, use the **async** SDK with an `async def` callback
> (the SDK awaits it without holding the GIL), or keep a sync callback fast by
> returning an already-cached token.

See [`examples/sync_example_federated.py`](examples/sync_example_federated.py) for a complete example.

### Custom headers provider

For advanced cases you can supply your own `HeadersProvider` via
`headers_provider=`; see the [`HeadersProvider`](#headersprovider) reference.

## Persistent Streams

A persistent (Eos) stream records its identity and committed offset on the server. After a
restart, you resume the same stream by its `stream_id`, and delivery into Delta stays
exactly-once.

Persistent streams support JSON and protobuf records. They authenticate with OAuth or a
custom headers provider. They do not support Avro or external-IdP federation.

> **Experimental.** The persistent stream API in the Rust core is in development. It can
> change in a minor release.

```python
import os

from zerobus.sdk.sync import ZerobusSdk

sdk = ZerobusSdk(host, unity_catalog_url)
builder = sdk.persistent_stream_builder().table("catalog.schema.events").oauth(client_id, client_secret).json()

# First run: create a stream. The server assigns the stream_id.
# Later runs: resume the stream with the saved stream_id.
saved_id = os.getenv("ZEROBUS_PERSISTENT_STREAM_ID")
if saved_id:
    stream = builder.resume(saved_id)  # use the same table and record format
    print(stream.resume_watermark)  # offset the server had committed, or None if none
else:
    stream = builder.build()
    print(f"Save this stream_id to resume later: {stream.stream_id}")

for record in records:
    stream.ingest_record_offset(record)
stream.flush()
stream.close()  # ends this connection only; the stream stays available to resume
```

Run the program once with no stream ID. Save the printed `stream_id`. After a restart, run it
again with the saved ID. `resume_watermark` is `None` when the server has committed nothing
yet. This is always the case for a new stream, and it can also be the case after `resume()`.

In the async SDK, `build()` and `resume()` are coroutines, and the stream methods are awaited.

### Recovering after a failure

Save the `stream_id` as soon as `build()` returns, before you ingest any record. If the
process stops before you save it, you cannot resume the stream.

A persistent stream has no `get_unacked_records()` and no `recreate_stream()`. To recover,
follow these steps:

1. Call `resume(stream_id)` with the same table and record format.
2. Read `resume_watermark`. This is the last offset the server committed, or `None` if it
   committed nothing.
3. Send again every record that your source holds after that offset. If `resume_watermark` is
   `None`, send again every record.

The SDK does not store your records. Your application must know which of its records
match which offsets, for example by storing its own source position with each batch.

Use one process per `stream_id`. The SDK does not check whether another client has
resumed the same stream.

The builder has these setters: `table`, `oauth`, `headers_provider`, `json`,
`compiled_proto`, `recovery`, `recovery_timeout_ms`, `recovery_backoff_ms`,
`recovery_retries`, `server_lack_of_ack_timeout_ms`, `flush_timeout_ms`,
`max_inflight_requests`, `max_ingest_payload_bytes`, `ack_callback`, and
`callback_max_wait_time_ms`. Unset values use the SDK defaults.

`PersistentStream` has `ingest_record_offset()`, `ingest_records_offset()`,
`wait_for_offset()`, `flush()`, and `close()`. It also has the properties `stream_id`,
`resume_watermark`, and `is_closed`. Unlike `ZerobusStream`, it has no `ingest_record()`,
no `*_nowait()` methods, and no `get_unacked_*()` methods.

See `examples/sync_example_persistent.py` and `examples/async_example_persistent.py`. Both examples read the optional `ZEROBUS_PERSISTENT_STREAM_ID` environment variable.

## Configuration

Configure stream behavior by passing a `StreamConfigurationOptions` object to `create_stream()`:

```python
from zerobus.sdk.shared import AckCallback, StreamConfigurationOptions

class MyCallback(AckCallback):
    def on_ack(self, offset: int):
        print(f"Acknowledged offset: {offset}")

    def on_error(self, offset: int, error_message: str):
        print(f"Error at offset {offset}: {error_message}")

options = StreamConfigurationOptions(
    max_inflight_records=10000,
    recovery=True,
    ack_callback=MyCallback()
)

stream = sdk.create_stream(client_id, client_secret, table_properties, options)
```

### Available Options

The record format is inferred from `TableProperties`: a Protobuf descriptor selects Protobuf,
an `avro_schema` selects Avro, and neither selects JSON. `record_type` defaults to
`RecordType.UNSPECIFIED` (infer from `TableProperties`); a non-`UNSPECIFIED` value does not
select the format but must match the inferred one, otherwise stream creation raises `ValueError`.

| Option                           | Type            | Default            | Description                                                                                                          |
| -------------------------------- | --------------- | ------------------ | -------------------------------------------------------------------------------------------------------------------- |
| `record_type`                    | `RecordType`    | `RecordType.UNSPECIFIED` | Optional; when not `UNSPECIFIED` it must match the format inferred from `TableProperties`                       |
| `max_inflight_records`           | `int`           | `1000000`          | Maximum number of unacknowledged records                                                                             |
| `recovery`                       | `bool`          | `True`             | Enable automatic stream recovery                                                                                     |
| `recovery_timeout_ms`            | `int`           | `15000`            | Timeout for recovery operations (ms)                                                                                 |
| `recovery_backoff_ms`            | `int`           | `2000`             | Delay between recovery attempts (ms)                                                                                 |
| `recovery_retries`               | `int`           | `4`                | Maximum number of recovery attempts                                                                                  |
| `flush_timeout_ms`               | `int`           | `300000`           | Timeout for flush operations (ms)                                                                                    |
| `server_lack_of_ack_timeout_ms`  | `int`           | `60000`            | Server acknowledgment timeout (ms)                                                                                   |
| `stream_paused_max_wait_time_ms` | `Optional[int]` | `None`             | Max wait during graceful stream close. `None` = full server duration, `0` = immediate, `x` = min(x, server_duration) |
| `callback_max_wait_time_ms`      | `Optional[int]` | `5000`             | Max wait for callbacks after `close()`. `None` = wait forever                                                        |
| `ack_callback`                   | `AckCallback`   | `None`             | Callback invoked once per successfully queued ingest submission that later acknowledges or fails                     |

## Error Handling

The SDK raises two categories of exception. Handle them differently:

- **Retriable** (`ZerobusException`): transient conditions such as network issues or
  temporary server errors. Safe to retry, ideally with backoff. The SDK's built-in
  recovery already handles many of these for you.
- **Non-retriable** (`NonRetriableException`): fatal conditions such as invalid
  credentials or a missing table. Retrying won't help. Fix the underlying problem.

`NonRetriableException` is a subclass of `ZerobusException`, so a bare
`except ZerobusException` still catches both.

```python
from zerobus.sdk.shared import ZerobusException, NonRetriableException

try:
    stream.ingest_record_offset(record)
except NonRetriableException as e:
    # Fatal: do not retry; fix the cause (credentials, table, schema)
    raise
except ZerobusException as e:
    # Transient: retry with backoff
    ...
```

## Handling Stream Failures

The SDK automatically handles retries for transient errors. Enqueue, flush, and close
failures surface as `ZerobusException` or `NonRetriableException`. `get_unacked_records()`
and `recreate_stream()` succeed only after the stream has already closed, which a terminal
failure does. An enqueue failure leaves the stream active, so those calls fail; raise
the original error and keep the stream. `recreate_stream()` re-queues records that were
already accepted; it does not retry a payload that failed to enqueue.

```python
from zerobus.sdk.shared import ZerobusException

try:
    for i in range(10000):
        stream.ingest_record_offset(record)
    stream.flush()
except ZerobusException as e:
    print(f"Ingestion failed: {e}")
    try:
        unacked = list(stream.get_unacked_records())
    except ZerobusException:
        raise e
    print(f"{len(unacked)} previously queued records were unacknowledged.")
    try:
        new_stream = sdk.recreate_stream(stream)
        try:
            new_stream.flush()
        finally:
            new_stream.close()
    except ZerobusException:
        raise e
else:
    stream.close()
```

Use `get_unacked_batches()` to inspect the original batch grouping after the stream closes:

```python
unacked_batches = list(stream.get_unacked_batches())
print(f"{len(unacked_batches)} batches remain unacknowledged")
```

**Decoding unacked records:**

- **JSON mode**: `json.loads(record_bytes.decode('utf-8'))`
- **Protobuf mode**: `YourMessage.FromString(record_bytes)`

## Performance Tips

The reliable bulk path is `ingest_records_offset()` plus one `flush()`. That call
amortizes the Python-to-Rust crossing and returns an offset after the batch is queued.
For single records, use `ingest_record_offset()` in a loop and `flush()` once. Ingest
calls queue immediately and the SDK acknowledges records in the background, so a single
`flush()` confirms everything queued so far. The ack watermark is monotonic, so if you
want a durability checkpoint mid-stream, waiting on the last offset returned confirms
every prior record. In async code, an [`AckCallback`](#ackcallback) tracks durability
without blocking. Calling `wait_for_offset()` after every record in a tight loop limits
throughput to one record per round-trip, so save it for confirming a specific record.

`ingest_record_nowait()` and `ingest_records_nowait()` spawn detached tasks and discard
enqueue errors. `flush()` can complete before those tasks allocate offsets, so they are
not a safe durability path. Prefer the offset APIs.

| Method                     | Throughput | Use case                                                                                          |
| -------------------------- | ---------- | ------------------------------------------------------------------------------------------------- |
| `ingest_records_offset()`  | Highest    | Recommended bulk path: queue a batch, then `flush()` once                                         |
| `ingest_record_offset()`   | Medium     | Recommended for single records: ingest in a loop, then `flush()` once                             |
| `ingest_record()`          | Low        | Deprecated; prefer offset-based APIs                                                              |
| `ingest_record_nowait()`   | Unsafe     | Detached fire-and-forget; enqueue errors can be lost and are not synchronized with `flush()`      |
| `ingest_records_nowait()`  | Unsafe     | Detached batch fire-and-forget; same durability caveats as `ingest_record_nowait()`               |

Idiomatic flow:

```python
async def ingest_all(stream, records):
    await stream.ingest_records_offset(records)     # queues the batch, no round-trip
    await stream.flush()                            # one wait for everything
```

**Confirming a specific record** (waiting on the last offset confirms all prior records):

```python
async def ingest_and_confirm(stream, records):
    offset = None
    for record in records:
        offset = await stream.ingest_record_offset(record)
    if offset is not None:
        await stream.wait_for_offset(offset)        # confirm the run before continuing
```

## API Reference

### `ZerobusSdk`

Main entry point. Sync: `from zerobus.sdk.sync import ZerobusSdk` / Async: `from zerobus.sdk.aio import ZerobusSdk`

```python
sdk = ZerobusSdk(
    host="https://<workspace>.zerobus.<region>.cloud.databricks.com",
    unity_catalog_url="https://<workspace-host>",
    application_name="my-app/1.0",
    connection_per_stream=True,
)
```

`application_name` is optional; when set it is appended to the `user-agent` header on gRPC requests to the Zerobus server (not on the OAuth token requests to the login service). It follows the `"<product>/<version>"` convention (e.g. `my-app/1.0`).

`connection_per_stream` gives every JSON/protobuf stream a dedicated gRPC
connection by default. Set it to `False` to share one HTTP/2 connection across
streams. Arrow Flight streams are unaffected.

HTTP/2 multiplexes logical streams over one TCP connection. On high-throughput
workloads over the public internet, packet loss and TCP retransmissions can
cause head-of-line blocking across every stream on that connection, reducing
aggregate throughput. Dedicated connections isolate that loss. For many
smaller, low-throughput streams, set `connection_per_stream=False`; shared
multiplexing is recommended there to reduce connection overhead.

```python
# Sync
stream = sdk.create_stream(client_id, client_secret, table_properties, options=None, headers_provider=None)
# Async
async def create_async_stream(sdk):
    return await sdk.create_stream(client_id, client_secret, table_properties, options=None, headers_provider=None)
```

### `ZerobusStream`

**Single record ingestion:**

| Method                         | Sync                     | Async                | Notes                                                                 |
| ------------------------------ | ------------------------ | -------------------- | --------------------------------------------------------------------- |
| `ingest_record_offset(record)` | `→ int`                  | `await → int`        | Recommended for single records; returns offset after queueing         |
| `ingest_record(record)`        | `→ RecordAcknowledgment` | `await → Awaitable`  | Deprecated since v0.3.0                                               |
| `ingest_record_nowait(record)` | `→ None`                 | `→ None` (not async) | Detached fire-and-forget; enqueue errors are not synchronized with `flush()` |

**Batch ingestion:**

| Method                           | Sync     | Async                | Notes                                                                 |
| -------------------------------- | -------- | -------------------- | --------------------------------------------------------------------- |
| `ingest_records_offset(records)` | `→ int`  | `await → int`        | Recommended bulk path; returns the batch's final offset               |
| `ingest_records_nowait(records)` | `→ None` | `→ None` (not async) | Detached fire-and-forget; same durability caveats as `ingest_record_nowait()` |

**Accepted record types:**

- **JSON mode**: `dict` (SDK serializes) or `str` (pre-serialized JSON)
- **Protobuf mode**: `Message` object (SDK serializes) or `bytes` (pre-serialized)

**Offset tracking** (use to confirm a specific record before continuing; for bulk
durability, ingest in a loop and `flush()` once):

```python
# Sync
offset = stream.ingest_record_offset(record)
# ... do other work ...
stream.wait_for_offset(offset)  # Block until durably written

# Async
async def confirm_async(stream, record):
    offset = await stream.ingest_record_offset(record)
    # ... do other work ...
    await stream.wait_for_offset(offset)  # Block until durably written
```

Acks are ordered, so waiting on the last offset returned confirms all prior records too.

**Stream management:**

```python
# Sync
stream.flush()   # Wait for all pending records to be acknowledged
stream.close()   # Flush and close gracefully (always call in finally)

# Async
async def close_async(stream):
    await stream.flush()
    await stream.close()
```

**Unacknowledged records:**

```python
# Sync
records = stream.get_unacked_records()   # Iterator[bytes]
batches = stream.get_unacked_batches()  # Iterator[List[bytes]]

# Async
async def get_unacked_async(stream):
    records = await stream.get_unacked_records()
    batches = await stream.get_unacked_batches()
    return records, batches
```

### `TableProperties`

```python
# JSON mode
TableProperties("catalog.schema.table")

# Protobuf mode
TableProperties("catalog.schema.table", descriptor_proto=MyMessage.DESCRIPTOR)

# Avro mode (beta) — mutually exclusive with descriptor_proto
TableProperties("catalog.schema.table", avro_schema=json_schema)
```

### `StreamConfigurationOptions`

See [Configuration](#configuration) for full parameter list.

### `AckCallback`

```python
from zerobus.sdk.shared import AckCallback

class MyCallback(AckCallback):
    def on_ack(self, offset: int) -> None:
        # Called once for each acknowledged single-record or batch submission
        pass

    def on_error(self, offset: int, error_message: str) -> None:
        # Called once when a single-record or batch submission encounters an error
        pass
```

`close()` waits at most `callback_max_wait_time_ms` (default 5000 ms) for
in-flight callbacks. A callback per queued submission is not guaranteed if that
budget expires.

### `HeadersProvider`

For custom authentication (e.g. custom token providers), implement `HeadersProvider` and pass it to `create_stream()`. Must include both `authorization` and `x-databricks-zerobus-table-name` headers. You may also override the optional `invalidate(self)` hook to drop cached auth state after the server rejects a token; the SDK calls it so a fresh token is fetched on the next `get_headers()`. See [`examples/`](examples/) for implementation details.

### `RecordAcknowledgment` (Sync only, deprecated)

```text
ack.wait_for_ack(timeout_sec=None)  # Block until acknowledged
ack.is_done() -> bool
```

### Exceptions

- `ZerobusException(message, cause=None)` - Base exception; retryable SDK failures are raised as this type
- `NonRetriableException(message, cause=None)` - Subclass for fatal errors (`ZerobusError::is_retryable()` is false)

## Debugging

The SDK uses Rust's `tracing` framework. Control log levels via `RUST_LOG`:

```bash
export RUST_LOG=info           # Default
export RUST_LOG=debug          # Detailed debugging
export RUST_LOG=trace          # Very verbose
export RUST_LOG=zerobus_sdk=debug  # Only SDK components
```

## Building from Source

Building from source requires the Rust toolchain 1.88 or newer (install from [rustup.rs](https://rustup.rs/)).

```bash
git clone https://github.com/databricks/zerobus-sdk.git
cd zerobus-sdk/python
make dev    # Set up venv and install in editable mode
make test   # Run tests
make build  # Build release wheel
```

For development workflows and detailed instructions, see [CONTRIBUTING.md](CONTRIBUTING.md).

## Community and Contributing

We are keen to hear feedback. Please [file issues](https://github.com/databricks/zerobus-sdk/issues).

See [CONTRIBUTING.md](CONTRIBUTING.md) for development setup and contribution guidelines.

## License

This project is licensed under the Apache License 2.0. See [LICENSE](LICENSE) for the full text.

