ChromaDB

pip install netra-sdk chromadb

npm install netra-sdk chromadb

from netra import Netra
import chromadb
import os

# Initialize Netra
Netra.init(
    headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}",
    trace_content=True
)

# Create ChromaDB client - automatically traced
client = chromadb.Client()
collection = client.get_or_create_collection(name="my_collection")

# Add documents
collection.add(
    ids=["id1", "id2"],
    documents=["Document 1", "Document 2"],
    metadatas=[{"source": "web"}, {"source": "book"}]
)

import { Netra } from "netra-sdk";
import { ChromaClient } from "chromadb";

// Initialize Netra
await Netra.init({
  headers: `x-api-key=${process.env.NETRA_API_KEY}`,
  traceContent: true
});

// Create ChromaDB client - automatically traced
const client = new ChromaClient();
const collection = await client.getOrCreateCollection({ name: "my_collection" });

// Add documents
await collection.add({
  ids: ["id1", "id2"],
  documents: ["Document 1", "Document 2"],
  metadatas: [{ source: "web" }, { source: "book" }]
});

from netra.decorators import task
from netra import SpanWrapper

@task()
def create_collection(client, name: str):
    span = SpanWrapper("chroma-create-collection", {
        "collection.name": name
    }).start()
    
    collection = client.create_collection(name=name)
    span.set_attribute("collection.id", collection.id)
    span.end()
    
    return collection

import { task, SpanWrapper } from "netra-sdk";

@task()
async function createCollection(client: ChromaClient, name: string) {
  const span = new SpanWrapper("chroma-create-collection", {
    "collection.name": name
  }).start();
  
  const collection = await client.createCollection({ name });
  span.setAttribute("collection.id", collection.id);
  span.end();
  
  return collection;
}

from netra.decorators import task
from netra import SpanWrapper, ActionModel

@task()
def add_documents(collection, documents: list[str]):
    span = SpanWrapper("chroma-add-documents", {
        "documents.count": len(documents)
    }).start()
    
    collection.add(
        ids=[f"doc{i}" for i in range(len(documents))],
        documents=documents
    )
    
    span.set_action([ActionModel(
        action="insert",
        action_type="database.insert",
        success=True,
        affected_records=[{"id": f"doc{i}"} for i in range(len(documents))],
        metadata={"collection": collection.name}
    )])
    span.set_attribute("status", "success")
    span.end()

import { task, SpanWrapper, ActionModel } from "netra-sdk";

@task()
async function addDocuments(collection: any, documents: string[]) {
  const span = new SpanWrapper("chroma-add-documents", {
    "documents.count": documents.length
  }).start();
  
  await collection.add({
    ids: documents.map((_, i) => `doc${i}`),
    documents: documents
  });
  
  span.setAction([{
    action: "insert",
    action_type: "database.insert",
    success: true,
    affected_records: documents.map((_, i) => ({ id: `doc${i}` })),
    metadata: { collection: collection.name }
  }]);
  span.setAttribute("status", "success");
  span.end();
}

from netra.decorators import workflow
from netra import SpanWrapper

@workflow()
def search_similar(collection, query: str, n: int = 5):
    span = SpanWrapper("chroma-search", {
        "query": query,
        "n_results": n
    }).start()
    
    results = collection.query(
        query_texts=[query],
        n_results=n
    )
    
    span.set_attribute("results.count", len(results["ids"][0]))
    span.end()
    
    return results

import { workflow, SpanWrapper } from "netra-sdk";

@workflow()
async function searchSimilar(collection: any, query: string, n: number = 5) {
  const span = new SpanWrapper("chroma-search", {
    "query": query,
    "n_results": n
  }).start();
  
  const results = await collection.query({
    queryTexts: [query],
    nResults: n
  });
  
  span.setAttribute("results.count", results.ids[0].length);
  span.end();
  
  return results;
}

from netra.decorators import task
from netra import SpanWrapper
import json

@task()
def filter_search(collection, query: str, filter: dict):
    span = SpanWrapper("chroma-filter-search", {
        "query": query,
        "filter": json.dumps(filter)
    }).start()
    
    results = collection.query(
        query_texts=[query],
        where=filter
    )
    
    span.set_attribute("results.count", len(results["ids"][0]))
    span.end()
    
    return results

import { task, SpanWrapper } from "netra-sdk";

@task()
async function filterSearch(collection: any, query: string, filter: any) {
  const span = new SpanWrapper("chroma-filter-search", {
    "query": query,
    "filter": JSON.stringify(filter)
  }).start();
  
  const results = await collection.query({
    queryTexts: [query],
    where: filter
  });
  
  span.setAttribute("results.count", results.ids[0].length);
  span.end();
  
  return results;
}