Weaviate

pip install netra-sdk weaviate-client

npm install netra-sdk weaviate-ts-client

from netra import Netra
import weaviate
import os

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

# Create Weaviate client - automatically traced
client = weaviate.Client(
    url=os.environ.get('WEAVIATE_URL'),
    auth_client_secret=weaviate.AuthApiKey(os.environ.get('WEAVIATE_API_KEY'))
)

# Query data
result = client.query.get("Article", ["title", "content"]).do()

import { Netra } from "netra-sdk";
import weaviate from "weaviate-ts-client";

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

// Create Weaviate client - automatically traced
const client = weaviate.client({
  scheme: "https",
  host: process.env.WEAVIATE_URL,
  apiKey: new weaviate.ApiKey(process.env.WEAVIATE_API_KEY)
});

// Query data
const result = await client.graphql
  .get()
  .withClassName("Article")
  .withFields("title content")
  .do();

from netra.decorators import task
from netra import SpanWrapper

@task()
def create_schema(client, class_name: str):
    span = SpanWrapper("weaviate-create-schema", {
        "class.name": class_name
    }).start()
    
    schema = {
        "class": class_name,
        "properties": [
            {"name": "title", "dataType": ["text"]},
            {"name": "content", "dataType": ["text"]}
        ]
    }
    
    client.schema.create_class(schema)
    span.end()

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

@task()
async function createSchema(client: any, className: string) {
  const span = new SpanWrapper("weaviate-create-schema", {
    "class.name": className
  }).start();
  
  const schema = {
    class: className,
    properties: [
      {
        name: "title",
        dataType: ["text"]
      },
      {
        name: "content",
        dataType: ["text"]
      }
    ]
  };
  
  await client.schema.classCreator().withClass(schema).do();
  span.end();
}

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

@task()
def add_object(client, class_name: str, properties: dict):
    span = SpanWrapper("weaviate-add-object", {
        "class.name": class_name
    }).start()
    
    result = client.data_object.create(
        data_object=properties,
        class_name=class_name
    )
    
    span.set_action([ActionModel(
        action="insert",
        action_type="database.insert",
        success=True,
        affected_records=[{"id": result}],
        metadata={"class": class_name}
    )])
    span.set_attribute("object.id", result)
    span.end()
    
    return result

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

@task()
async function addObject(client: any, className: string, properties: any) {
  const span = new SpanWrapper("weaviate-add-object", {
    "class.name": className
  }).start();
  
  const result = await client.data
    .creator()
    .withClassName(className)
    .withProperties(properties)
    .do();
  
  span.setAction([{
    action: "insert",
    action_type: "database.insert",
    success: true,
    affected_records: [{ id: result.id }],
    metadata: { class: className }
  }]);
  span.setAttribute("object.id", result.id);
  span.end();
  
  return result;
}

from netra.decorators import workflow
from netra import SpanWrapper

@workflow()
def semantic_search(client, class_name: str, query: str, limit: int = 5):
    span = SpanWrapper("weaviate-search", {
        "class.name": class_name,
        "query": query,
        "limit": limit
    }).start()
    
    results = (
        client.query
        .get(class_name, ["title", "content"])
        .with_near_text({"concepts": [query]})
        .with_limit(limit)
        .with_additional(["distance"])
        .do()
    )
    
    count = len(results.get("data", {}).get("Get", {}).get(class_name, []))
    span.set_attribute("results.count", count)
    span.end()
    
    return results

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

@workflow()
async function semanticSearch(client: any, className: string, query: string, limit: number = 5) {
  const span = new SpanWrapper("weaviate-search", {
    "class.name": className,
    "query": query,
    "limit": limit
  }).start();
  
  const results = await client.graphql
    .get()
    .withClassName(className)
    .withFields("title content _additional { distance }")
    .withNearText({ concepts: [query] })
    .withLimit(limit)
    .do();
  
  const count = results.data.Get[className]?.length || 0;
  span.setAttribute("results.count", count);
  span.end();
  
  return results;
}

from netra.decorators import task
from netra import SpanWrapper

@task()
def hybrid_search(client, class_name: str, query: str):
    span = SpanWrapper("weaviate-hybrid-search", {
        "class.name": class_name,
        "query": query
    }).start()
    
    results = (
        client.query
        .get(class_name, ["title", "content"])
        .with_hybrid(query=query, alpha=0.5)
        .do()
    )
    
    count = len(results.get("data", {}).get("Get", {}).get(class_name, []))
    span.set_attribute("results.count", count)
    span.end()
    
    return results

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

@task()
async function hybridSearch(client: any, className: string, query: string) {
  const span = new SpanWrapper("weaviate-hybrid-search", {
    "class.name": className,
    "query": query
  }).start();
  
  const results = await client.graphql
    .get()
    .withClassName(className)
    .withFields("title content")
    .withHybrid({ query, alpha: 0.5 })
    .do();
  
  const count = results.data.Get[className]?.length || 0;
  span.setAttribute("results.count", count);
  span.end();
  
  return results;
}