Weaviate

pip install netra-sdk weaviate-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()

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()

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

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

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