vdb-python 1.1.0

Python SDK for the VectorDB REST API

vdb-python

Python SDK for the VectorDB REST API.

Sync and async clients, full API coverage, CLI tool included.

Installation

pip install vdb-python

Quick Start

from vectordb_client import VectorDBClient

client = VectorDBClient(
    base_url="https://your-vectordb.onrender.com",
    api_key="your-api-key",
)

# Create a collection
client.collections.create("articles", dim=384, distance_metric="cosine")

# Store a vector
client.vectors.upsert(
    "articles",
    external_id="doc-1",
    vector=[0.1, 0.2, 0.9, ...],  # must match dim=384
    metadata={"title": "Hello World", "author": "Alice"},
)

# Search
results = client.search.search("articles", vector=query_vector, k=5)
for r in results:
    print(r.external_id, r.score, r.metadata)

Async Client

import asyncio
from vectordb_client import AsyncVectorDBClient

async def main():
    async with AsyncVectorDBClient(
        base_url="https://your-vectordb.onrender.com",
        api_key="your-api-key",
    ) as client:
        await client.collections.create("articles", dim=384)
        results = await client.search.search("articles", vector=query_vector, k=5)
        for r in results:
            print(r.external_id, r.score)

asyncio.run(main())

---

Auth

# Register a new user (returns user + api_key)
result = client.auth.register("[e-mail]", "password123")
api_key = result["api_key"]["key"]

# Login
result = client.auth.login("[e-mail]", "password123")

---

Collections

# Create
col = client.collections.create("articles", dim=384, distance_metric="cosine")
# distance_metric options: "cosine" (default), "l2", "ip"

# List
cols = client.collections.list()
for c in cols:
    print(c.name, c.dim, c.vector_count)

# Get
col = client.collections.get("articles")

# Update description
client.collections.update("articles", description="News articles corpus")

# Export all vectors
export = client.collections.export("articles", limit=10000)
for v in export.vectors:
    print(v.external_id, v.vector[:5])

# Delete
client.collections.delete("articles")

---

Vectors

# Upsert (insert or update)
result = client.vectors.upsert(
    "articles",
    external_id="doc-1",
    vector=[0.1, 0.2, ...],
    metadata={"title": "Hello", "tags": ["ml", "nlp"]},
)
print(result.status)  # "inserted" or "updated"

# Upsert with auto-embedding (server-side, if embedding provider configured)
result = client.vectors.upsert("articles", external_id="doc-1", text="Hello world")

# Bulk upsert
result = client.vectors.bulk_upsert("articles", items=[
    {"external_id": "doc-1", "vector": [...], "metadata": {"title": "A"}},
    {"external_id": "doc-2", "vector": [...], "metadata": {"title": "B"}},
    {"external_id": "doc-3", "text": "auto-embed this"},
])
print(f"Inserted: {len(result.inserted)}, Updated: {len(result.updated)}")

# Fetch a single vector
vec = client.vectors.get_vector("articles", "doc-1")

# Fetch multiple vectors by ID
vecs = client.vectors.batch_fetch("articles", ids=["doc-1", "doc-2", "doc-3"])

# Scroll through all vectors (cursor-based pagination)
page = client.vectors.scroll("articles", limit=100)
while page.has_more:
    for v in page.vectors:
        print(v["external_id"])
    page = client.vectors.scroll("articles", limit=100, cursor=page.next_cursor)

# Scroll with metadata filter
page = client.vectors.scroll("articles", limit=50, filters={"author": "Alice"})

# Delete one
client.vectors.delete("articles", "doc-1")

# Batch delete
client.vectors.delete_batch("articles", ids=["doc-1", "doc-2", "doc-3"])

---

Search

# KNN search
results = client.search.search("articles", vector=query_vector, k=10)
for r in results:
    print(r.external_id, r.score, r.metadata)

# Search with metadata filters
results = client.search.search(
    "articles",
    vector=query_vector,
    k=10,
    filters={"author": "Alice", "year": 2024},
)

# Search with pagination
results = client.search.search("articles", vector=query_vector, k=10, offset=20)

# Search by text (server-side embedding)
results = client.search.search("articles", text="machine learning", k=5)

# Recommendations — similar to a stored vector, excludes itself
recs = client.search.recommend("articles", external_id="doc-1", k=5)

# Cosine similarity between two stored vectors
score = client.search.similarity("articles", id1="doc-1", id2="doc-2")
print(f"Similarity: {score:.4f}")

# Rerank candidates against a query vector
reranked = client.search.rerank(
    "articles",
    query_vector=query_vector,
    candidates=["doc-1", "doc-2", "doc-3", "doc-4"],
)
for r in reranked:
    print(r.external_id, r.score)

# Hybrid search — vector + keyword via Reciprocal Rank Fusion
results = client.search.hybrid_search(
    "articles",
    query_text="transformer models attention",
    vector=query_vector,
    k=10,
    alpha=0.7,  # 0 = keyword only, 1 = vector only, 0.5 = balanced
)

# Bulk search — multiple queries in one request
results = client.search.bulk_search("articles", queries=[
    {"vector": query1, "k": 5},
    {"vector": query2, "k": 5, "filters": {"author": "Alice"}},
])
# results[0] = hits for query1, results[1] = hits for query2

---

RAG — Documents & Query

# Upload a document (auto-chunked and stored)
result = client.documents.upload("articles", file_path="paper.txt")
print(f"Created {result.chunks_created} chunks, doc_id={result.document_id}")

# Semantic search returning text chunks
result = client.query.query(
    query="What is attention mechanism?",
    collection_name="articles",
    top_k=5,
)
for chunk in result.results:
    print(chunk.score, chunk.text)

# LLM-powered answer with source citations
answer = client.query.ask(
    query="Explain transformers in simple terms",
    collection="articles",
    k=5,
)
print(answer.answer)
for src in answer.sources:
    print(src.external_id, src.score)

---

GraphRAG (Pro/Scale tier)

# Check graph status
status = client.graph.status("articles")
print(f"Entities: {status.entity_count}, Edges: {status.edge_count}")
print(f"Pending jobs: {status.pending_jobs}")

# Search the knowledge graph
results = client.graph.search("articles", query="transformer architecture", top_k=10)
for entity in results.entities:
    print(entity.entity, entity.entity_type, entity.score)
    for rel in entity.relations:
        print(f"  → {rel['relation']} → {rel['target']}")

# Find shortest path between two entities
path = client.graph.path("articles", source="BERT", target="GPT-4", max_hops=3)
for route in path.paths:
    for step in route:
        print(f"{step.source} --[{step.relation}]--> {step.target}")

# Louvain community detection (Scale tier)
summary = client.graph.summarize("articles", resolution=1.0)
for community in summary.communities:
    print(f"Community {community.id}: {community.entities[:3]}...")

# Full GraphRAG answer (Scale tier)
answer = client.graph.ask(
    "articles",
    query="How does BERT relate to GPT?",
    top_k_entities=5,
    max_hops=2,
)
print(answer.answer)
print(f"Used {len(answer.entities_used)} entities, {answer.paths_used} paths")

# Hybrid vector + graph answer (Scale tier)
answer = client.graph.hybrid_ask(
    "articles",
    query="What are attention mechanisms?",
    top_k=5,
    alpha=0.5,
)
print(answer.answer)
for src in answer.sources:
    print(src.external_id, src.source, src.score)  # source: "vector" or "graph"

# Update graph config
client.graph.config("articles", model="gpt-4o-mini", chunk_size=512)

---

Usage & Quotas

# Current usage
usage = client.usage.get_current()
print(f"Tier: {usage.tier}")
print(f"Requests: {usage.request_count}/{usage.max_requests}")
print(f"Vectors: {usage.vector_count}/{usage.max_vectors}")
for warning in usage.warnings:
    print(f"Warning: {warning}")

# Usage history
history = client.usage.get_history(days=30)

# Admin: upgrade a user's tier
client.usage.update_user_tier(user_id=42, tier="pro")

# Admin: trigger cleanup of expired data
client.usage.trigger_cleanup()

---

API Keys

# Create a key
key = client.keys.create("my-app", role="readwrite", expires_in_days=90)
print(key.key)  # shown only once

# List all keys
keys = client.keys.list()
for k in keys:
    print(k.id, k.name, k.role, k.is_active)

# Get one key
key = client.keys.get(key_id=3)

# Update
client.keys.update(key_id=3, name="renamed", role="readonly")

# Revoke / restore
client.keys.revoke(key_id=3)
client.keys.restore(key_id=3)

# Rotate (generates new key value)
new_key = client.keys.rotate(key_id=3)
print(new_key.key)  # new value, shown once

# Per-key usage stats
stats = client.keys.get_usage(key_id=3)
print(stats.total_requests, stats.last_24h, stats.by_endpoint)

# Overall usage summary
summary = client.keys.get_usage_summary()

# Delete
client.keys.delete(key_id=3)

---

Observability

# Health check
health = client.observability.health()
print(health.status)            # "ok"
print(health.total_vectors)
print(health.total_collections)

# Prometheus metrics (raw text)
metrics = client.observability.metrics()
print(metrics)

---

Error Handling

from vectordb_client import (
    VectorDBError,
    NotFoundError,
    AlreadyExistsError,
    DimensionMismatchError,
    AuthenticationError,
    RateLimitError,
)

try:
    client.collections.create("articles", dim=384)
except AlreadyExistsError:
    print("Collection already exists")
except DimensionMismatchError as e:
    print(f"Wrong vector size: {e}")
except AuthenticationError:
    print("Invalid or expired API key")
except RateLimitError:
    print("Too many requests — back off and retry")
except VectorDBError as e:
    print(f"API error {e.status_code}: {e}")

---

CLI

export VECTORDB_URL=https://your-vectordb.onrender.com
export VECTORDB_API_KEY=your-key

# Collections
vdb collections list
vdb collections create articles --dim 384 --metric cosine
vdb collections get articles
vdb collections delete articles

# Search
vdb search articles '[0.1, 0.2, 0.3]' --k 5

# JSON output
vdb -o json collections list

---

Configuration

Parameter	Default	Description
base_url	—	VectorDB server URL
api_key	—	API key (x-api-key header)
timeout	30	Request timeout in seconds

---

Requirements

• Python 3.10+
• requests — sync client
• httpx — async client

Links

• GitHub
• Full API Docs
• PyPI