Python SDK

Python SDK

Installation

pip install zeppelin-python

# From source
cd clients/python && pip install -e ".[dev]"

Quick Start

from zeppelin import ZeppelinClient, Vector

client = ZeppelinClient(base_url="http://localhost:8080")

# Create namespace
ns = client.create_namespace("my_vectors", dimensions=384, distance_metric="cosine")

# Upsert vectors
count = client.upsert_vectors("my_vectors", [
    Vector(id="vec-1", values=[0.1, 0.2, ...], attributes={"color": "red"}),
    Vector(id="vec-2", values=[0.3, 0.4, ...], attributes={"color": "blue"}),
])

# Query
response = client.query("my_vectors", vector=[0.1, 0.2, ...], top_k=10)
for result in response.results:
    print(f"{result.id}: {result.score}")

Client Options

client = ZeppelinClient(
    base_url="http://localhost:8080",  # Server URL
    timeout=30.0,                      # Request timeout (seconds)
    headers={"Authorization": "Bearer token"},  # Custom headers
)

Both sync and async clients support context managers:

with ZeppelinClient() as client:
    # ...

async with AsyncZeppelinClient() as client:
    # ...

Async Client

from zeppelin import AsyncZeppelinClient

async def main():
    client = AsyncZeppelinClient(base_url="http://localhost:8080")
    ns = await client.create_namespace("my_vectors", dimensions=384)
    await client.upsert_vectors("my_vectors", vectors)
    response = await client.query("my_vectors", vector=[0.1, ...], top_k=10)

Namespace Operations

# Create
ns = client.create_namespace("my_ns", dimensions=384, distance_metric="cosine")

# Create with FTS
from zeppelin import FtsFieldConfig
ns = client.create_namespace(
    "articles",
    dimensions=384,
    full_text_search={
        "title": FtsFieldConfig(k1=1.5, b=0.75),
        "content": FtsFieldConfig(),
    },
)

# List
namespaces = client.list_namespaces()

# Get
ns = client.get_namespace("my_ns")

# Delete
client.delete_namespace("my_ns")

Vector Operations

from zeppelin import Vector

# Upsert
count = client.upsert_vectors("my_ns", [
    Vector(
        id="vec-1",
        values=[0.1, 0.2, 0.3],
        attributes={
            "category": "product",
            "price": 29.99,
            "tags": ["electronics", "gadgets"],
        },
    ),
])

# Delete
count = client.delete_vectors("my_ns", ["vec-1", "vec-2"])

Querying

Vector search

response = client.query(
    "my_ns",
    vector=[0.1, 0.2, 0.3],
    top_k=10,
    nprobe=20,
    consistency="strong",
)

for result in response.results:
    print(f"{result.id}: score={result.score}, attrs={result.attributes}")

print(f"Scanned {response.scanned_fragments} fragments, {response.scanned_segments} segments")

BM25 search

from zeppelin import RankBy

response = client.query(
    "articles",
    rank_by=RankBy.bm25("content", "vector search engine"),
    top_k=10,
)

Weighted multi-field BM25

response = client.query(
    "articles",
    rank_by=RankBy.sum(
        RankBy.product(2.0, RankBy.bm25("title", "vector search")),
        RankBy.bm25("content", "vector search"),
    ),
    top_k=10,
)

Prefix search (autocomplete)

response = client.query(
    "articles",
    rank_by=RankBy.bm25("title", "vec"),
    last_as_prefix=True,
    top_k=5,
)

Filter Builder

from zeppelin import Filter

# Equality
f = Filter.eq("color", "red")

# Inequality
f = Filter.not_eq("status", "deleted")

# Range
f = Filter.range("price", gte=10.0, lte=100.0)

# Set membership
f = Filter.in_("category", ["electronics", "books"])
f = Filter.not_in("category", ["spam"])

# List contains
f = Filter.contains("tags", "rust")

# FTS token filters
f = Filter.contains_all_tokens("content", ["rust", "programming"])
f = Filter.contains_token_sequence("content", ["vector", "search"])

# Logical operators
f = Filter.and_(
    Filter.eq("color", "red"),
    Filter.range("price", lte=50.0),
)
f = Filter.or_(
    Filter.eq("color", "red"),
    Filter.eq("color", "blue"),
)
f = Filter.not_(Filter.eq("archived", True))

Query with filters

response = client.query(
    "my_ns",
    vector=[0.1, 0.2, 0.3],
    top_k=10,
    filter=Filter.and_(
        Filter.eq("category", "product"),
        Filter.range("price", gte=10.0, lte=100.0),
    ),
)

RankBy Builder

from zeppelin import RankBy

# Single-field BM25
expr = RankBy.bm25("content", "search query")

# Sum of multiple fields
expr = RankBy.sum(
    RankBy.bm25("title", "query"),
    RankBy.bm25("content", "query"),
)

# Max across fields
expr = RankBy.max(
    RankBy.bm25("title", "query"),
    RankBy.bm25("content", "query"),
)

# Weighted
expr = RankBy.product(2.0, RankBy.bm25("title", "query"))

# Combined
expr = RankBy.sum(
    RankBy.product(2.0, RankBy.bm25("title", "query")),
    RankBy.bm25("content", "query"),
)

Error Handling

from zeppelin import ZeppelinError, NotFoundError, ValidationError, ConflictError, ServerError

try:
    client.get_namespace("nonexistent")
except NotFoundError as e:
    print(f"Not found: {e.message} (status: {e.status_code})")  # 404
except ValidationError as e:
    print(f"Bad request: {e.message}")  # 400
except ConflictError as e:
    print(f"Conflict: {e.message}")  # 409
except ServerError as e:
    print(f"Server error: {e.message}")  # 5xx
except ZeppelinError as e:
    print(f"Error: {e.message} (status: {e.status_code})")

Health Checks

health = client.health()       # {"status": "ok"}
ready = client.ready()         # {"status": "ready", "s3_connected": True}