Pinecone

pip install netra-sdk pinecone-client

from netra import Netra
from pinecone import Pinecone
import os

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

# Create Pinecone client - automatically traced
pc = Pinecone(api_key=os.environ.get('PINECONE_API_KEY'))
index = pc.Index("my-index")

# Upsert vectors
index.upsert(vectors=[{
    "id": "vec1",
    "values": [0.1, 0.2, 0.3],
    "metadata": {"text": "Sample document"}
}])

from netra import task, SpanWrapper

@task()
def create_index(pc: Pinecone, name: str, dimension: int):
    span = SpanWrapper("pinecone-create-index", {
        "index.name": name,
        "index.dimension": dimension
    }).start()
    
    pc.create_index(
        name=name,
        dimension=dimension,
        metric="cosine",
        spec={"serverless": {"cloud": "aws", "region": "us-west-2"}}
    )
    
    span.end()

from netra import task, SpanWrapper, ActionModel

@task()
def upsert_vectors(index, vectors: list):
    span = SpanWrapper("pinecone-upsert", {
        "vectors.count": len(vectors)
    }).start()
    
    result = index.upsert(vectors=vectors)
    
    span.set_action([ActionModel(
        action="upsert",
        action_type="database.upsert",
        success=True,
        affected_records=[{"id": v["id"]} for v in vectors],
        metadata={"index": index._index_name, "upserted": result["upserted_count"]}
    )])
    span.set_attribute("upserted.count", result["upserted_count"])
    span.end()
    
    return result

from netra import workflow, SpanWrapper

@workflow()
def search_vectors(index, query: list[float], top_k: int = 5):
    span = SpanWrapper("pinecone-search", {
        "query.dimension": len(query),
        "top_k": top_k
    }).start()
    
    results = index.query(
        vector=query,
        top_k=top_k,
        include_metadata=True
    )
    
    span.set_attribute("results.count", len(results["matches"]))
    span.end()
    
    return results

from netra import task, SpanWrapper

@task()
def query_namespace(index, namespace: str, query: list[float]):
    span = SpanWrapper("pinecone-namespace-query", {
        "namespace": namespace
    }).start()
    
    results = index.query(
        vector=query,
        top_k=10,
        namespace=namespace
    )
    
    span.set_attribute("results.count", len(results["matches"]))
    span.end()
    
    return results