Pinecone

pip install netra-sdk pinecone-client

npm install netra-sdk @pinecone-database/pinecone

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"}
}])

import { Netra } from "netra-sdk";
import { Pinecone } from "@pinecone-database/pinecone";

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

// Create Pinecone client - automatically traced
const pinecone = new Pinecone({
  apiKey: process.env.PINECONE_API_KEY
});

const index = pinecone.index("my-index");

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

from netra.decorators import task
from netra import 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()

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

@task()
async function createIndex(pinecone: Pinecone, name: string, dimension: number) {
  const span = new SpanWrapper("pinecone-create-index", {
    "index.name": name,
    "index.dimension": dimension
  }).start();
  
  await pinecone.createIndex({
    name,
    dimension,
    metric: "cosine",
    spec: {
      serverless: {
        cloud: "aws",
        region: "us-west-2"
      }
    }
  });
  
  span.end();
}

from netra.decorators import task
from netra import 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

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

@task()
async function upsertVectors(index: any, vectors: any[]) {
  const span = new SpanWrapper("pinecone-upsert", {
    "vectors.count": vectors.length
  }).start();
  
  const result = await index.upsert(vectors);
  
  span.setAction([{
    action: "upsert",
    action_type: "database.upsert",
    success: true,
    affected_records: vectors.map(v => ({ id: v.id })),
    metadata: { index: index.name, upserted: result.upsertedCount }
  }]);
  span.setAttribute("upserted.count", result.upsertedCount);
  span.end();
  
  return result;
}

from netra.decorators import workflow
from netra import 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

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

@workflow()
async function searchVectors(index: any, query: number[], topK: number = 5) {
  const span = new SpanWrapper("pinecone-search", {
    "query.dimension": query.length,
    "top_k": topK
  }).start();
  
  const results = await index.query({
    vector: query,
    topK,
    includeMetadata: true
  });
  
  span.setAttribute("results.count", results.matches.length);
  span.end();
  
  return results;
}

from netra.decorators import task
from netra import 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

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

@task()
async function queryNamespace(index: any, namespace: string, query: number[]) {
  const span = new SpanWrapper("pinecone-namespace-query", {
    "namespace": namespace
  }).start();
  
  const results = await index.namespace(namespace).query({
    vector: query,
    topK: 10
  });
  
  span.setAttribute("results.count", results.matches.length);
  span.end();
  
  return results;
}