### Setup Development Environment

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Create and activate a virtual environment, installing the package in development mode with all dev dependencies. This command also builds the Rust extension.

```bash
make dev
```

--------------------------------

### Install Zerobus SDK from Source

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Commands to install Rust, clone the SDK repository, and install the package in editable mode using pip. Requires the Rust toolchain.

```bash
# Install Rust (if not already installed)
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

# Clone and install
git clone https://github.com/databricks/zerobus-sdk-py.git
cd zerobus-sdk-py
pip install -e .
```

--------------------------------

### Install Databricks Zerobus Ingest SDK

Source: https://context7.com/databricks/zerobus-sdk-py/llms.txt

Install the SDK using pip. Building from source requires the Rust toolchain.

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

```bash
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
git clone https://github.com/databricks/zerobus-sdk-py.git
cd zerobus-sdk-py
pip install -e .
```

--------------------------------

### Zerobus SDK Authentication Setup

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Demonstrates the basic setup for initializing the Zerobus SDK using OAuth 2.0 Client Credentials for authentication. This snippet shows the initial SDK and TableProperties configuration.

```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(table_name, record_pb2.AirQuality.DESCRIPTOR)

```

--------------------------------

### Run Zerobus SDK Examples

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/examples/README.md

Execute the example Python scripts to test synchronous and asynchronous ingestion for both protobuf and JSON record types.

```bash
# Synchronous examples (blocking I/O)
python examples/sync_example_proto.py     # Protobuf
python examples/sync_example_json.py      # JSON

# Asynchronous examples (non-blocking I/O)
python examples/async_example_proto.py    # Protobuf
python examples/async_example_json.py     # JSON
```

--------------------------------

### Install Zerobus SDK Dependencies

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/examples/README.md

Install the Zerobus SDK and its dependencies in editable mode. This makes the SDK available for use in your project.

```bash
pip install -e .
```

--------------------------------

### Example Google-Style Docstring

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Use Google-style docstrings for public APIs, including arguments, return values, exceptions, and examples.

```python
def ingest_record(self, record) -> RecordAcknowledgment:
    """
    Submits a single record for ingestion into the stream.

    This method may block if the maximum number of in-flight records
    has been reached.

    Args:
        record: The Protobuf message object to be ingested.

    Returns:
        RecordAcknowledgment: An object to wait on for the server's acknowledgment.

    Raises:
        ZerobusException: If the stream is not in a valid state for ingestion.

    Example:
        >>> record = AirQuality(device_name="sensor-1", temp=25)
        >>> ack = stream.ingest_record(record)
        >>> ack.wait_for_ack()
    """
```

--------------------------------

### Install Zerobus SDK via Pip

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Command to install the latest stable version of the Databricks Zerobus Ingest SDK using pip.

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

--------------------------------

### Generate .proto from Unity Catalog Schema (CLI)

Source: https://context7.com/databricks/zerobus-sdk-py/llms.txt

Use the `generate_proto` CLI tool to fetch a Unity Catalog table schema and emit a proto2 file. Install dependencies and then run the command, followed by compiling the generated file with `grpc_tools.protoc`.

```bash
# Install the dependency (only needed for this tool)
pip install "requests>=2.28.1,<3"

# Generate the .proto file
python -m zerobus.tools.generate_proto \
    --uc-endpoint  "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com" \
    --client-id    "my-service-principal-id" \
    --client-secret "my-service-principal-secret" \
    --table        "main.default.air_quality" \
    --proto-msg    "AirQuality" \
    --output       "air_quality.proto"

# Compile the generated file
pip install "grpcio-tools>=1.60.0,<2.0"
python -m grpc_tools.protoc --python_out=. --proto_path=. air_quality.proto

# Use in your code
import air_quality_pb2
from zerobus.sdk.shared import TableProperties
props = TableProperties("main.default.air_quality", air_quality_pb2.AirQuality.DESCRIPTOR)
```

--------------------------------

### Clone Zerobus SDK Repository

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/examples/README.md

Clone the Zerobus SDK repository to access the example applications. Navigate into the cloned directory.

```bash
git clone https://github.com/databricks/zerobus-sdk-py.git
cd zerobus-sdk-py
```

--------------------------------

### Run Linters

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Check your code for style guide violations and unused imports using pycodestyle and autoflake.

```bash
make lint
```

--------------------------------

### Set Multiple Log Targets with Different Levels

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Define different log levels for multiple components, for example, setting 'zerobus_sdk' to 'trace' and 'tokio' to 'info'.

```bash
export RUST_LOG=zerobus_sdk=trace,tokio=info
```

--------------------------------

### Example Commit Message

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Follow conventions for commit messages: present tense, imperative mood, 50-character limit for the first line, and reference issues.

```text
Add async stream creation example

- Add async_example.py demonstrating non-blocking ingestion
- Update README with async API documentation

Fixes #42
```

--------------------------------

### Asynchronous Data Ingestion with Protobuf

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

This asynchronous example is suitable for non-blocking data ingestion using Protocol Buffers. It requires an asyncio event loop and proper configuration of server endpoints and authentication. The SDK can serialize Message objects or accept pre-serialized bytes.

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

# Configure logging (optional but recommended)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

async def main():
    # Configuration
    # For AWS:
    server_endpoint = "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com"
    workspace_url = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com"
    # For Azure:
    # server_endpoint = "https://1234567890123456.zerobus.us-west-2.azuredatabricks.net"
    # workspace_url = "https://dbc-a1b2c3d4-e5f6.azuredatabricks.net"

    table_name = "main.default.air_quality"
    client_id = "your-service-principal-application-id"
    client_secret = "your-service-principal-secret"

    # Initialize SDK
    sdk = ZerobusSdk(server_endpoint, workspace_url)

    # Configure table properties with protobuf descriptor
    table_properties = TableProperties(table_name, record_pb2.AirQuality.DESCRIPTOR)

    # Create stream
    stream = await sdk.create_stream(client_id, client_secret, table_properties)

    try:
        # Ingest records
        for i in range(100):
            # Option 1: Pass a Message object (SDK serializes to bytes)
            record = record_pb2.AirQuality(
                device_name=f"sensor-{i % 10}",
                temp=20 + (i % 15),
                humidity=50 + (i % 40)
            )
            future = await stream.ingest_record(record)

            # Option 2: Pass pre-serialized bytes (client controls serialization)
            # serialized_bytes = record.SerializeToString()
            # future = await stream.ingest_record(serialized_bytes)

            # Optional: Wait for durability confirmation
            await future

            print(f"Ingested record {i + 1}")

        print("Successfully ingested 100 records!")
finally:
        await stream.close()

asyncio.run(main())
```

--------------------------------

### Build Rust Extension (Development Mode)

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Compile the Rust code in debug mode and install it directly into your virtual environment. Re-run this command after modifying Rust code.

```bash
make develop-rust
```

--------------------------------

### Workflow After Modifying Rust Code

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

After modifying Rust code, rebuild and install the Rust extension, then test the Python bindings.

```bash
make develop-rust   # Rebuild and install Rust extension
make test          # Test Python bindings
```

--------------------------------

### Async Ingestion with Offset and Fire-and-Forget

Source: https://context7.com/databricks/zerobus-sdk-py/llms.txt

Async equivalents of sync ingestion methods. Note that ingest_record_nowait and ingest_records_nowait are synchronous even in the async SDK and should not be awaited. This example demonstrates collecting offsets concurrently and fire-and-forget ingestion.

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

async def ingest_demo():
    server_endpoint = "YOUR_SERVER_ENDPOINT"
    workspace_url = "YOUR_WORKSPACE_URL"
    client_id = "YOUR_CLIENT_ID"
    client_secret = "YOUR_CLIENT_SECRET"

    sdk = ZerobusSdk(server_endpoint, workspace_url)
    stream = await sdk.create_stream(
        client_id, client_secret,
        TableProperties("main.default.sensors"),
        StreamConfigurationOptions(record_type=RecordType.JSON, max_inflight_records=100_000),
    )
    try:
        # Collect offsets concurrently
        offsets = []
        for i in range(200):
            offset = await stream.ingest_record_offset({"id": i, "v": i * 2})
            offsets.append(offset)

        # Fire-and-forget: NOT awaited
        for i in range(200, 1_000):
            stream.ingest_record_nowait({"id": i, "v": i * 2})

        await stream.wait_for_offset(offsets[-1])
        await stream.flush()
        print("Done")
    finally:
        await stream.close()

asyncio.run(ingest_demo())
```

--------------------------------

### Protobuf Ingestion (Sync)

Source: https://context7.com/databricks/zerobus-sdk-py/llms.txt

Pass a compiled protobuf Message object or pre-serialised bytes directly to any ingest method. The SDK serialises Message objects internally. Ensure grpcio-tools is installed.

```python
# Compile the schema first:
#   pip install "grpcio-tools>=1.60.0,<2.0"
```

--------------------------------

### Build Rust Extension (Release Mode)

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Create an optimized wheel in the `target/wheels/` directory. Use this for performance testing or creating release builds.

```bash
make build-rust
```

--------------------------------

### Initialize Synchronous Zerobus SDK

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Instantiate the synchronous Zerobus SDK client. Requires Zerobus and Unity Catalog endpoint URLs.

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

server_endpoint = "<workspace-id>.zerobus.<region>.cloud.databricks.com"
unity_catalog_endpoint = "<your workspace URL>"

sdk = ZerobusSdk(server_endpoint, unity_catalog_endpoint)
```

--------------------------------

### Initialize Asynchronous Zerobus SDK

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Instantiate the asynchronous Zerobus SDK client. Requires Zerobus and Unity Catalog endpoint URLs.

```python
from zerobus.sdk.aio import ZerobusSdk

server_endpoint = "<workspace-id>.zerobus.<region>.cloud.databricks.com"
unity_catalog_endpoint = "<your workspace URL>"

sdk = ZerobusSdk(server_endpoint, unity_catalog_endpoint)
```

--------------------------------

### Creating a Release Build

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Build an optimized wheel for release by running the `build-rust` make target. The wheel will be located in the `target/wheels/` directory.

```bash
make build-rust    # Build optimized wheel
# Wheel will be in target/wheels/
```

--------------------------------

### Initialize ZerobusSdk (Sync)

Source: https://context7.com/databricks/zerobus-sdk-py/llms.txt

Initialize the synchronous Zerobus SDK client. TLS is enabled by default; disable for local testing. Ensure environment variables for server endpoint, workspace URL, client ID, and client secret are set.

```python
import os
from zerobus.sdk.sync import ZerobusSdk, ZerobusStream
from zerobus.sdk.shared import TableProperties, StreamConfigurationOptions, RecordType
from zerobus import ZerobusException, NonRetriableException

SERVER_ENDPOINT   = os.environ["ZEROBUS_SERVER_ENDPOINT"]   # e.g. https://<id>.zerobus.us-west-2.cloud.databricks.com
WORKSPACE_URL     = os.environ["DATABRICKS_WORKSPACE_URL"]  # e.g. https://<workspace>.cloud.databricks.com
CLIENT_ID         = os.environ["DATABRICKS_CLIENT_ID"]
CLIENT_SECRET     = os.environ["DATABRICKS_CLIENT_SECRET"]
TABLE             = "main.default.air_quality"


sdk = ZerobusSdk(SERVER_ENDPOINT, WORKSPACE_URL)

table_props = TableProperties(TABLE)
options = StreamConfigurationOptions(
    record_type=RecordType.JSON,
    max_inflight_records=10_000,
    recovery=True,
    recovery_retries=3,
)

stream: ZerobusStream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_props, options)
# Optional custom headers provider instead of OAuth:
# stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_props, options, 
#                            headers_provider=my_provider)

try:
    stream.flush()
finally:
    stream.close()

# Recreate a closed stream with the same configuration:
new_stream = sdk.recreate_stream(stream)
new_stream.close()
```

--------------------------------

### Initialize ZerobusSdk (Async)

Source: https://context7.com/databricks/zerobus-sdk-py/llms.txt

Initialize the asynchronous Zerobus SDK client. Stream creation and management methods are coroutines. Fire-and-forget ingest methods are not awaitable.

```python
import asyncio, os
from zerobus.sdk.aio import ZerobusSdk
from zerobus.sdk.shared import TableProperties, StreamConfigurationOptions, RecordType

async def main():
    sdk = ZerobusSdk(
        os.environ["ZEROBUS_SERVER_ENDPOINT"],
        os.environ["DATABRICKS_WORKSPACE_URL"],
    )

    table_props = TableProperties("main.default.air_quality")
    options = StreamConfigurationOptions(record_type=RecordType.JSON, max_inflight_records=50_000)

    stream = await sdk.create_stream(
        os.environ["DATABRICKS_CLIENT_ID"],
        os.environ["DATABRICKS_CLIENT_SECRET"],
        table_props,
        options,
    )
    try:
        await stream.flush()
    finally:
        await stream.close()

asyncio.run(main())
```

--------------------------------

### Ingest Protobuf Record with Offset Tracking

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/examples/README.md

Demonstrates creating a protobuf record and ingesting it using `ingest_record_offset()`. This method provides offset tracking for better performance.

```python
# Create protobuf record
record = record_pb2.AirQuality(device_name="sensor-1", temp=25, humidity=60)
table_properties = TableProperties(TABLE_NAME, record_pb2.AirQuality.DESCRIPTOR)
options = StreamConfigurationOptions(record_type=RecordType.PROTO)

# Recommended: Use ingest_record_offset() for better performance
offset = stream.ingest_record_offset(record)
```

--------------------------------

### ZerobusSdk Constructor (Synchronous)

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Initializes the synchronous Zerobus SDK client. Requires server and Unity Catalog endpoints.

```APIDOC
## ZerobusSdk Constructor (Synchronous)

### Description
Initializes the synchronous Zerobus SDK client. Requires server and Unity Catalog endpoints.

### Parameters
- `server_endpoint` (str) - The Zerobus gRPC endpoint (e.g., `<workspace-id>.zerobus.<region>.cloud.databricks.com` for AWS, or `<workspace-id>.zerobus.<region>.azuredatabricks.net` for Azure)
- `unity_catalog_endpoint` (str) - The Unity Catalog endpoint (your workspace URL)
```

--------------------------------

### Ingest Single Record and Get Offset

Source: https://context7.com/databricks/zerobus-sdk-py/llms.txt

Use ingest_record_offset for single-record ingestion, returning an integer offset immediately without blocking on server acknowledgment. Supports both JSON (dict/str) and Protobuf (bytes/Message) record types.

```python
import json
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import TableProperties, StreamConfigurationOptions, RecordType

# Assuming SERVER_ENDPOINT and WORKSPACE_URL are defined elsewhere
sdk = ZerobusSdk(SERVER_ENDPOINT, WORKSPACE_URL)
stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET,
                           TableProperties("main.default.sensors"),
                           StreamConfigurationOptions(record_type=RecordType.JSON))
try:
    offsets = []
    for i in range(50):
        record = {"device": f"sensor-{i}", "temp": 20 + i % 15, "humidity": 50 + i % 40}
        # Pass a dict — the SDK serialises it to JSON internally
        offset = stream.ingest_record_offset(record)
        offsets.append(offset)

        # OR pass a pre-serialised string for full control:
        # offset = stream.ingest_record_offset(json.dumps(record))

    # Wait for a specific offset to be durably acknowledged
    stream.wait_for_offset(offsets[-1])
    print(f"All records up to offset {offsets[-1]} acknowledged")
    stream.flush()
finally:
    stream.close()
```

--------------------------------

### Get Unacknowledged Batches

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Retrieves a list of unacknowledged batches, where each batch is a list of records as raw bytes. Useful for implementing batch retry logic. Records are returned in their serialized form; refer to `get_unacked_records()` for decoding instructions.

```python
async def get_unacked_batches() -> List[List[bytes]]
```

--------------------------------

### Get Unacknowledged Records

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Retrieves a list of raw bytes representing records that have been ingested but not yet acknowledged by the server. Use for recovery and monitoring. Decode JSON with `json.loads(record.decode('utf-8'))` or deserialize Protobuf with `YourMessage.FromString(record)`.

```python
async def get_unacked_records() -> List[bytes]
```

--------------------------------

### Initialize Zerobus SDK with JSON Serialization (Python)

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Python code to initialize the Zerobus SDK for synchronous operations using JSON serialization. Configure server endpoint, workspace URL, and authentication details.

```python
import json
import logging
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import RecordType, StreamConfigurationOptions, TableProperties

# Configure logging (optional but recommended)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

# Configuration
# For AWS:
server_endpoint = "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com"
workspace_url = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com"
# For Azure:
# server_endpoint = "https://1234567890123456.zerobus.us-west-2.azuredatabricks.net"
# workspace_url = "https://dbc-a1b2c3d4-e5f6.azuredatabricks.net"

table_name = "main.default.air_quality"
client_id = "your-service-principal-application-id"
client_secret = "your-service-principal-secret"

# Initialize SDK
sdk = ZerobusSdk(server_endpoint, workspace_url)

# Configure table properties
table_properties = TableProperties(table_name)

# Configure stream with JSON record type
options = StreamConfigurationOptions(record_type=RecordType.JSON)

```

--------------------------------

### Get Unacknowledged Batches (Sync)

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Returns a list of unacknowledged batches, where each batch is a list of records as raw bytes. These batches have been sent but not yet acknowledged by the server. Records are returned in their serialized form. Useful for batch retry logic.

```python
def get_unacked_batches() -> List[List[bytes]]
```

--------------------------------

### Generate Protobuf Schema from Unity Catalog

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Automatically generate a protobuf schema file from a Unity Catalog table. This tool simplifies schema creation by inferring types from an existing table.

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

--------------------------------

### ZerobusSdk Constructor (Asynchronous)

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Initializes the asynchronous Zerobus SDK client. Requires server and Unity Catalog endpoints.

```APIDOC
## ZerobusSdk Constructor (Asynchronous)

### Description
Initializes the asynchronous Zerobus SDK client. Requires server and Unity Catalog endpoints.

### Parameters
- `server_endpoint` (str) - The Zerobus gRPC endpoint (e.g., `<workspace-id>.zerobus.<region>.cloud.databricks.com` for AWS, or `<workspace-id>.zerobus.<region>.azuredatabricks.net` for Azure)
- `unity_catalog_endpoint` (str) - The Unity Catalog endpoint (your workspace URL)
```

--------------------------------

### Get Unacknowledged Records (Sync)

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Returns a list of unacknowledged records as raw bytes. These records have been ingested but not yet acknowledged by the server. Records are returned in their serialized form and may need decoding based on the ingestion mode (JSON or Protobuf). Useful for recovery and monitoring.

```python
def get_unacked_records() -> List[bytes]
```

--------------------------------

### Python SDK Architecture Overview

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Illustrates the layered architecture of the Python SDK, highlighting the interaction between the Python application, the Python SDK wrapper, and the Rust core implementation.

```text
┌─────────────────────────────────────────┐
│         Python Application Code         │
│  (Your code using the Python SDK API)  │
└─────────────────────────────────────────┘
                    │
                    ▼
┌─────────────────────────────────────────┐
│       Python SDK (Thin Wrapper)         │
│    • API compatibility layer            │
│    • Python types & error handling      │
└─────────────────────────────────────────┘
                    │
                    ▼ (PyO3 bindings)
┌─────────────────────────────────────────┐
│         Rust Core Implementation        │
│    • gRPC communication                 │
│    • OAuth 2.0 authentication           │
│    • Stream management & recovery       │
│    • Protocol encoding/decoding         │
└─────────────────────────────────────────┘
```

--------------------------------

### Activate Virtual Environment

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Activate the Python virtual environment. Use the appropriate command for your operating system.

```bash
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
```

--------------------------------

### Configure Stream with Options

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Configure stream behavior by passing a StreamConfigurationOptions object to create_stream(). All options are optional; defaults are used if not specified.

```python
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import StreamConfigurationOptions, RecordType, TableProperties



sdk = ZerobusSdk(server_endpoint, workspace_url)
table_properties = TableProperties(table_name)

# Optional: Create a custom callback class
class MyCallback(AckCallback):
    def on_ack(self, offset: int):
        print(f"Ack: {offset}")

# Create options with custom configuration
options = StreamConfigurationOptions(
    record_type=RecordType.JSON,
    max_inflight_records=10000,
    recovery=True,
    recovery_timeout_ms=20000,
    ack_callback=MyCallback()  # Optional - can be None
)

# Pass options when creating the stream
stream = sdk.create_stream(
    client_id,
    client_secret,
    table_properties,
    options  # <-- Configuration options passed here
)
```

--------------------------------

### Ingest JSON Record with Offset Tracking

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/examples/README.md

Demonstrates creating a JSON record as a Python dictionary and ingesting it using `ingest_record_offset()`. This method is recommended for performance with offset tracking.

```python
# Create JSON record
record_dict = {"device_name": "sensor-1", "temp": 25, "humidity": 60}
table_properties = TableProperties(TABLE_NAME)
options = StreamConfigurationOptions(record_type=RecordType.JSON)

# Recommended: Use ingest_record_offset() for better performance
offset = stream.ingest_record_offset(record_dict)
```

--------------------------------

### Daily Development Workflow

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Run these commands for daily development involving Python changes: format code, check for issues, and run tests.

```bash
make fmt        # Format your code
make lint       # Check for issues
make test       # Run tests
```

--------------------------------

### Open Coverage Report

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Open the generated HTML coverage report in your browser. Use the appropriate command for your operating system.

```bash
open htmlcov/index.html  # macOS
xdg-open htmlcov/index.html  # Linux
```

--------------------------------

### Instantiate TableProperties for Protobuf Mode

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Creates a `TableProperties` object for a table when using Protobuf mode. Requires both the fully qualified table name and the Protobuf message descriptor.

```python
# Protobuf mode (default)
table_properties = TableProperties("catalog.schema.table", record_pb2.MyMessage.DESCRIPTOR)
```

--------------------------------

### Format Code

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Run the code formatter and other code quality tools to ensure consistent style. This includes Black for formatting, autoflake for removing unused imports, and isort for sorting imports.

```bash
make fmt
```

--------------------------------

### ZerobusSdk (Sync) — zerobus.sdk.sync.ZerobusSdk

Source: https://context7.com/databricks/zerobus-sdk-py/llms.txt

Main entry point for synchronous Zerobus ingestion. Wraps the Rust core SDK and exposes stream creation and recreation methods. TLS is enabled by default.

```APIDOC
## ZerobusSdk (Sync) — zerobus.sdk.sync.ZerobusSdk

### Description
Main entry point for synchronous Zerobus ingestion. Wraps the Rust core SDK and exposes `create_stream`, `create_stream_with_headers_provider`, and `recreate_stream`. TLS is enabled by default; call `set_use_tls(False)` only for local testing against a mock server.

### Method Signature
`ZerobusSdk.create_stream(client_id: str, client_secret: str, table_properties: TableProperties, stream_options: StreamConfigurationOptions, headers_provider: Optional[Callable] = None) -> ZerobusStream`
`ZerobusSdk.recreate_stream(stream: ZerobusStream) -> ZerobusStream`

### Parameters
#### `create_stream` Parameters
- **client_id** (str) - Required - The client ID for authentication.
- **client_secret** (str) - Required - The client secret for authentication.
- **table_properties** (TableProperties) - Required - Properties of the target table.
- **stream_options** (StreamConfigurationOptions) - Required - Configuration options for the stream.
- **headers_provider** (Optional[Callable]) - Optional - A custom function to provide headers.

#### `recreate_stream` Parameters
- **stream** (ZerobusStream) - Required - The stream to recreate.

### Request Example
```python
import os
from zerobus.sdk.sync import ZerobusSdk, ZerobusStream
from zerobus.sdk.shared import TableProperties, StreamConfigurationOptions, RecordType

SERVER_ENDPOINT   = os.environ["ZEROBUS_SERVER_ENDPOINT"]
WORKSPACE_URL     = os.environ["DATABRICKS_WORKSPACE_URL"]
CLIENT_ID         = os.environ["DATABRICKS_CLIENT_ID"]
CLIENT_SECRET     = os.environ["DATABRICKS_CLIENT_SECRET"]
TABLE             = "main.default.air_quality"

sdk = ZerobusSdk(SERVER_ENDPOINT, WORKSPACE_URL)

table_props = TableProperties(TABLE)
options = StreamConfigurationOptions(
    record_type=RecordType.JSON,
    max_inflight_records=10_000,
    recovery=True,
    recovery_retries=3,
)

stream: ZerobusStream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_props, options)

try:
    stream.flush()
finally:
    stream.close()

# Recreate a closed stream with the same configuration:
new_stream = sdk.recreate_stream(stream)
new_stream.close()
```

### Response
#### Success Response (`create_stream`, `recreate_stream`)
- **ZerobusStream** - An instance of `ZerobusStream` for ingesting data.
```

--------------------------------

### Push to Fork

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Push your local feature branch to your fork on GitHub.

```bash
git push origin feature/your-feature-name
```

--------------------------------

### Configure Databricks and Zerobus Credentials

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/examples/README.md

Set environment variables for Databricks authentication and Zerobus endpoint configuration. Ensure you use the correct values for your cloud provider (AWS or Azure).

```bash
export DATABRICKS_CLIENT_ID="your-service-principal-application-id"
export DATABRICKS_CLIENT_SECRET="your-service-principal-secret"
# For AWS:
export ZEROBUS_SERVER_ENDPOINT="https://workspace-id.zerobus.region.cloud.databricks.com"
export DATABRICKS_WORKSPACE_URL="https://your-workspace.cloud.databricks.com"
# For Azure:
# export ZEROBUS_SERVER_ENDPOINT="https://workspace-id.zerobus.region.azuredatabricks.net"
# export DATABRICKS_WORKSPACE_URL="https://your-workspace.azuredatabricks.net"
export ZEROBUS_TABLE_NAME="catalog.schema.table"
```

--------------------------------

### Create Stream with OAuth

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Use this to create a stream with OAuth 2.0 authentication. The SDK handles authentication and TLS connections by default.

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

--------------------------------

### Create JSON Record for Ingestion

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/examples/README.md

Prepare a JSON record as a string for ingestion into Zerobus. Ensure the JSON structure conforms to the target Databricks table schema.

```python
import json

json_record = json.dumps({"field1": "value1", "field2": 123})
```

--------------------------------

### Ingest Protobuf Records with Zerobus SDK

Source: https://context7.com/databricks/zerobus-sdk-py/llms.txt

Ingests 1,000 protobuf records into a Zerobus stream. Demonstrates passing Message objects directly for serialization by the SDK. Ensure the proto descriptor is correctly passed.

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

# Assuming SERVER_ENDPOINT, WORKSPACE_URL, CLIENT_ID, CLIENT_SECRET are defined
sdk = ZerobusSdk(SERVER_ENDPOINT, WORKSPACE_URL)
table_props = TableProperties(
    "main.default.air_quality",
    record_pb2.AirQuality.DESCRIPTOR,   # pass Descriptor object
)
stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_props)
try:
    for i in range(1_000):
        record = record_pb2.AirQuality(
            device_name=f"sensor-{i % 10}",
            temp=20 + i % 15,
            humidity=50 + i % 40,
        )
        # Option A: pass Message object (SDK serialises)
        offset = stream.ingest_record_offset(record)

        # Option B: pass pre-serialised bytes
        # offset = stream.ingest_record_offset(record.SerializeToString())

    stream.flush()
    print("1 000 protobuf records written")
finally:
    stream.close()
```

--------------------------------

### Synchronous Data Ingestion with Protobuf

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Use this snippet for synchronous data ingestion using Protocol Buffers. Ensure you have configured logging, server endpoints, and authentication details. The SDK handles serialization if a Message object is passed.

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

# Configure logging (optional but recommended)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

# Configuration
# For AWS:
server_endpoint = "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com"
workspace_url = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com"
# For Azure:
# server_endpoint = "https://1234567890123456.zerobus.us-west-2.azuredatabricks.net"
# workspace_url = "https://dbc-a1b2c3d4-e5f6.azuredatabricks.net"

table_name = "main.default.air_quality"
client_id = "your-service-principal-application-id"
client_secret = "your-service-principal-secret"

# Initialize SDK
sdk = ZerobusSdk(server_endpoint, workspace_url)

# Configure table properties with protobuf descriptor
table_properties = TableProperties(table_name, record_pb2.AirQuality.DESCRIPTOR)

# Create stream
stream = sdk.create_stream(client_id, client_secret, table_properties)

try:
    # Ingest records
    for i in range(100):
        # Option 1: Pass a Message object (SDK serializes to bytes)
        record = record_pb2.AirQuality(
            device_name=f"sensor-{i % 10}",
            temp=20 + (i % 15),
            humidity=50 + (i % 40)
        )
        ack = stream.ingest_record(record)

        # Option 2: Pass pre-serialized bytes (client controls serialization)
        # serialized_bytes = record.SerializeToString()
        # ack = stream.ingest_record(serialized_bytes)

        # Optional: Wait for durability confirmation
        ack.wait_for_ack()

        print(f"Ingested record {i + 1}")

    print("Successfully ingested 100 records!")
finally:
    stream.close()
```

--------------------------------

### Set Trace Log Level for Zerobus SDK

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Enable the most detailed trace logging for the Zerobus SDK by setting the RUST_LOG environment variable to 'trace'.

```bash
export RUST_LOG=trace
```

--------------------------------

### Commit Changes

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Stage all changes and commit them with a descriptive message.

```bash
git add .
git commit -m "Add feature: description of your changes"
```

--------------------------------

### Fire-and-Forget Protobuf Record Ingestion

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/examples/README.md

Shows how to ingest a protobuf record using `ingest_record_nowait()`. This method is suitable for maximum throughput as it does not wait for confirmation.

```python
# Or fire-and-forget for maximum throughput
stream.ingest_record_nowait(record)
```

--------------------------------

### ZerobusSdk (Async) — zerobus.sdk.aio.ZerobusSdk

Source: https://context7.com/databricks/zerobus-sdk-py/llms.txt

Asynchronous counterpart of the sync SDK. All stream-creation and stream-management methods are coroutines. Fire-and-forget ingest methods are not awaitable.

```APIDOC
## ZerobusSdk (Async) — zerobus.sdk.aio.ZerobusSdk

### Description
Asynchronous counterpart of the sync SDK. All stream-creation and stream-management methods are coroutines. `ingest_record_nowait` and `ingest_records_nowait` are intentionally **not** awaitable — call them without `await` for fire-and-forget throughput.

### Method Signature
`async ZerobusSdk.create_stream(client_id: str, client_secret: str, table_properties: TableProperties, stream_options: StreamConfigurationOptions) -> ZerobusStream`

### Parameters
#### `create_stream` Parameters
- **client_id** (str) - Required - The client ID for authentication.
- **client_secret** (str) - Required - The client secret for authentication.
- **table_properties** (TableProperties) - Required - Properties of the target table.
- **stream_options** (StreamConfigurationOptions) - Required - Configuration options for the stream.

### Request Example
```python
import asyncio, os
from zerobus.sdk.aio import ZerobusSdk
from zerobus.sdk.shared import TableProperties, StreamConfigurationOptions, RecordType

async def main():
    sdk = ZerobusSdk(
        os.environ["ZEROBUS_SERVER_ENDPOINT"],
        os.environ["DATABRICKS_WORKSPACE_URL"],
    )

    table_props = TableProperties("main.default.air_quality")
    options = StreamConfigurationOptions(record_type=RecordType.JSON, max_inflight_records=50_000)

    stream = await sdk.create_stream(
        os.environ["DATABRICKS_CLIENT_ID"],
        os.environ["DATABRICKS_CLIENT_SECRET"],
        table_props,
        options,
    )
    try:
        await stream.flush()
    finally:
        await stream.close()

asyncio.run(main())
```

### Response
#### Success Response (`create_stream`)
- **ZerobusStream** - An instance of `ZerobusStream` for ingesting data.
```

--------------------------------

### StreamConfigurationOptions Constructor

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Configure the behavior of a stream, including record serialization, in-flight limits, recovery settings, and timeouts.

```APIDOC
## StreamConfigurationOptions Constructor

### Description
Configuration options for stream behavior.

### Method
```python
StreamConfigurationOptions(
    record_type: RecordType = RecordType.PROTO,
    max_inflight_records: int = 50000,
    recovery: bool = True,
    recovery_timeout_ms: int = 15000,
    recovery_backoff_ms: int = 2000,
    recovery_retries: int = 3,
    flush_timeout_ms: int = 300000,
    server_lack_of_ack_timeout_ms: int = 60000,
    stream_paused_max_wait_time_ms: Optional[int] = None,
    callback_max_wait_time_ms: Optional[int] = 5000,
    ack_callback: AckCallback = None
)
```

### Parameters
#### Path Parameters
- `record_type` (RecordType) - Serialization format: `RecordType.PROTO` (default) or `RecordType.JSON`
- `max_inflight_records` (int) - Maximum number of unacknowledged records (default: 50000)
- `recovery` (bool) - Enable or disable automatic stream recovery (default: True)
- `recovery_timeout_ms` (int) - Recovery operation timeout in milliseconds (default: 15000)
- `recovery_backoff_ms` (int) - Delay between recovery attempts in milliseconds (default: 2000)
- `recovery_retries` (int) - Maximum number of recovery attempts (default: 3)
- `flush_timeout_ms` (int) - Flush operation timeout in milliseconds (default: 300000)
- `server_lack_of_ack_timeout_ms` (int) - Server acknowledgment timeout in milliseconds (default: 60000)
- `stream_paused_max_wait_time_ms` (Optional[int]) - Maximum time in milliseconds to wait during graceful stream close. When the server signals stream closure, the SDK can pause and wait for in-flight records to be acknowledged. `None` = wait for full server-specified duration (most graceful), `0` = immediate recovery, `x` = wait up to min(x, server_duration) milliseconds (default: None)
- `callback_max_wait_time_ms` (Optional[int]) - Maximum time in milliseconds to wait for callbacks to finish after calling `close()` on the stream. `None` = wait forever, `x` = wait up to x milliseconds (default: 5000)
- `ack_callback` (AckCallback) - Callback to be invoked when records are acknowledged or encounter errors. Must be a custom class extending `AckCallback` that implements `on_ack()` and optionally `on_error()` methods. (default: None)

### Request Example
```python
from zerobus.sdk.shared import StreamConfigurationOptions, RecordType, AckCallback

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

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

options = StreamConfigurationOptions(
    record_type=RecordType.JSON,
    max_inflight_records=10000,
    ack_callback=MyCallback()
)

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

--------------------------------

### Run Tests

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Execute the test suite using pytest to ensure changes do not break existing functionality. This also generates coverage reports.

```bash
make test
```

--------------------------------

### Generate Protobuf Python Code

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/examples/README.md

Use grpc_tools.protoc to generate Python code from a .proto schema file. Ensure the proto file is in the current directory or specify the correct proto_path.

```bash
python -m grpc_tools.protoc --python_out=. --proto_path=. your_schema.proto
```

--------------------------------

### Non-Blocking Protobuf Ingestion with Zerobus SDK

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Enables asynchronous ingestion of Protocol Buffer data using asyncio. Features include custom acknowledgment callbacks and configurable in-flight record limits. Requires protobuf descriptor.

```python
import asyncio
import logging
from zerobus.sdk.aio import ZerobusSdk
from zerobus.sdk.shared import TableProperties, StreamConfigurationOptions, AckCallback
import record_pb2

logging.basicConfig(level=logging.INFO)

async def main():
    # Create a custom callback class
    class MyCallback(AckCallback):
        def on_ack(self, offset: int):
            print(f"Acknowledged offset: {offset}")

    options = StreamConfigurationOptions(
        max_inflight_records=50000,
        ack_callback=MyCallback()
    )

    sdk = ZerobusSdk(server_endpoint, workspace_url)
    table_properties = TableProperties(table_name, record_pb2.AirQuality.DESCRIPTOR)
    stream = await sdk.create_stream(client_id, client_secret, table_properties, options)

    futures = []
    try:
        for i in range(100000):
            # Pass a Message object (SDK serializes) or pre-serialized bytes
            record = record_pb2.AirQuality(
                device_name=f"sensor-{i % 10}",
                temp=20 + i % 15,
                humidity=50 + i % 40
            )
            future = await stream.ingest_record(record)
            futures.append(future)

        await stream.flush()
        await asyncio.gather(*futures)
    finally:
        await stream.close()

asyncio.run(main())
```

--------------------------------

### Asynchronous Record Ingestion with Zerobus SDK

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Ingest records asynchronously using the aio Zerobus SDK. Supports passing dictionaries (SDK serializes) or pre-serialized JSON strings. Asynchronous operations require awaiting futures and closing the stream.

```python
import asyncio
import json
import logging
from zerobus.sdk.aio import ZerobusSdk
from zerobus.sdk.shared import RecordType, StreamConfigurationOptions, TableProperties

# Configure logging (optional but recommended)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

async def main():
    # Configuration
    # For AWS:
    server_endpoint = "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com"
    workspace_url = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com"
    # For Azure:
    # server_endpoint = "1234567890123456.zerobus.us-west-2.azuredatabricks.net"
    # workspace_url = "https://dbc-a1b2c3d4-e5f6.azuredatabricks.net"

    table_name = "main.default.air_quality"
    client_id = "your-service-principal-application-id"
    client_secret = "your-service-principal-secret"

    # Initialize SDK
    sdk = ZerobusSdk(server_endpoint, workspace_url)

    # Configure table properties
    table_properties = TableProperties(table_name)

    # Configure stream with JSON record type
    options = StreamConfigurationOptions(record_type=RecordType.JSON)

    # Create stream
    stream = await sdk.create_stream(client_id, client_secret, table_properties, options)

    try:
        # Ingest records
        for i in range(100):
            # Option 1: Pass a dict (SDK serializes to JSON)
            record_dict = {
                "device_name": f"sensor-{i % 10}",
                "temp": 20 + (i % 15),
                "humidity": 50 + (i % 40)
            }
            future = await stream.ingest_record(record_dict)

            # Option 2: Pass a pre-serialized JSON string (client controls serialization)
            # json_string = json.dumps(record_dict)
            # future = await stream.ingest_record(json_string)

            # Optional: Wait for durability confirmation
            await future

            print(f"Ingested record {i + 1}")

        print("Successfully ingested 100 records!")
    finally:
        await stream.close()

asyncio.run(main())
```

--------------------------------

### Create Feature Branch

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/CONTRIBUTING.md

Create a new branch for your feature or bug fix.

```bash
git checkout -b feature/your-feature-name
```

--------------------------------

### Blocking Protobuf Ingestion with Zerobus SDK

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Use this for synchronous ingestion of Protocol Buffer data. Requires a generated protobuf descriptor (e.g., record_pb2.AirQuality.DESCRIPTOR). Ensure the SDK and shared components are imported.

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

logging.basicConfig(level=logging.INFO)

sdk = ZerobusSdk(server_endpoint, workspace_url)
table_properties = TableProperties(table_name, record_pb2.AirQuality.DESCRIPTOR)
stream = sdk.create_stream(client_id, client_secret, table_properties)

try:
    for i in range(1000):
        # Pass a Message object (SDK serializes) or pre-serialized bytes
        record = record_pb2.AirQuality(
            device_name=f"sensor-{i}",
            temp=20 + i % 15,
            humidity=50 + i % 40
        )
        ack = stream.ingest_record(record)

        # Optional: Wait for durability confirmation
        ack.wait_for_ack()
finally:
    stream.close()
```

--------------------------------

### Compile Protocol Buffer Schema

Source: https://github.com/databricks/zerobus-sdk-py/blob/main/README.md

Compile a .proto file into Python code using grpc_tools.protoc. This command generates a Python file compatible with protobuf 6.x.

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