LanceDB

pip install netra-sdk lancedb

from netra import Netra
import lancedb
import os

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

# Connect to LanceDB - automatically traced
db = lancedb.connect("./lancedb")
table = db.create_table("my_table", [
    {"id": 1, "vector": [0.1, 0.2], "text": "Sample"}
])

from netra import task, SpanWrapper

@task()
def create_table(db, name: str, data: list):
    span = SpanWrapper("lancedb-create-table", {
        "table.name": name,
        "data.count": len(data)
    }).start()
    
    table = db.create_table(name, data)
    span.end()
    
    return table

from netra import task, SpanWrapper, ActionModel

@task()
def add_vectors(table, data: list):
    span = SpanWrapper("lancedb-add", {
        "vectors.count": len(data)
    }).start()
    
    table.add(data)
    
    span.set_action([ActionModel(
        action="insert",
        action_type="database.insert",
        success=True,
        affected_records=[{"id": str(d["id"])} for d in data],
        metadata={"table": table.name}
    )])
    span.set_attribute("status", "success")
    span.end()

from netra import workflow, SpanWrapper

@workflow()
def search_vectors(table, query: list[float], limit: int = 5):
    span = SpanWrapper("lancedb-search", {
        "query.size": len(query),
        "limit": limit
    }).start()
    
    results = table.search(query).limit(limit).to_list()
    
    span.set_attribute("results.count", len(results))
    span.end()
    
    return results

from netra import task, SpanWrapper

@task()
def filter_search(table, query: list[float], filter: str):
    span = SpanWrapper("lancedb-filter-search", {
        "filter": filter
    }).start()
    
    results = table.search(query).where(filter).limit(10).to_list()
    
    span.set_attribute("results.count", len(results))
    span.end()
    
    return results