Qdrant

pip install netra-sdk qdrant-client

from netra import Netra
from qdrant_client import QdrantClient
import os

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

# Create Qdrant client - automatically traced
client = QdrantClient(
    url=os.environ.get('QDRANT_URL'),
    api_key=os.environ.get('QDRANT_API_KEY')
)

# Create collection
client.create_collection(
    collection_name="my_collection",
    vectors_config={"size": 384, "distance": "Cosine"}
)

from netra import task, SpanWrapper

@task()
def create_collection(client: QdrantClient, name: str, size: int):
    span = SpanWrapper("qdrant-create-collection", {
        "collection.name": name,
        "vector.size": size
    }).start()
    
    client.create_collection(
        collection_name=name,
        vectors_config={"size": size, "distance": "Cosine"}
    )
    
    span.end()

from netra import task, SpanWrapper, ActionModel

@task()
def upsert_points(client: QdrantClient, collection: str, points: list):
    span = SpanWrapper("qdrant-upsert", {
        "collection": collection,
        "points.count": len(points)
    }).start()
    
    client.upsert(
        collection_name=collection,
        points=points
    )
    
    span.set_action([ActionModel(
        action="upsert",
        action_type="database.upsert",
        success=True,
        affected_records=[{"id": str(p.id)} for p in points],
        metadata={"collection": collection}
    )])
    span.set_attribute("status", "success")
    span.end()

from netra import workflow, SpanWrapper

@workflow()
def search_points(client: QdrantClient, collection: str, query: list[float], limit: int = 5):
    span = SpanWrapper("qdrant-search", {
        "collection": collection,
        "query.size": len(query),
        "limit": limit
    }).start()
    
    results = client.search(
        collection_name=collection,
        query_vector=query,
        limit=limit
    )
    
    span.set_attribute("results.count", len(results))
    span.end()
    
    return results

from netra import task, SpanWrapper
import json

@task()
def filter_search(client: QdrantClient, collection: str, query: list[float], filter: dict):
    span = SpanWrapper("qdrant-filter-search", {
        "collection": collection,
        "filter": json.dumps(filter)
    }).start()
    
    results = client.search(
        collection_name=collection,
        query_vector=query,
        query_filter=filter,
        limit=10
    )
    
    span.set_attribute("results.count", len(results))
    span.end()
    
    return results