Error handling

from actian_vectorai import (
    CollectionNotFoundError,
    TimeoutError,
    UnimplementedError,
    VectorAIClient,
    VectorAIError,
    get_retry_delay,
    is_retryable,
)

query = [0.1] * 128

with VectorAIClient("localhost:6574") as client:
    try:
        results = client.points.search("products", vector=query, limit=10)
    except CollectionNotFoundError as error:
        print(f"Collection '{error.collection_name}' not found")
    except UnimplementedError as error:
        print(f"Operation '{error.operation}' is not supported by this server")
    except TimeoutError:
        print("Request timed out; try increasing the per-call timeout")
    except VectorAIError as error:
        if is_retryable(error):
            delay = get_retry_delay(error, attempt=1)
            print(f"Transient error, retry after {delay:.1f}s")
        else:
            raise

from actian_vectorai import (
    CollectionExistsError,
    Distance,
    VectorAIClient,
    VectorParams,
)

with VectorAIClient("localhost:6574") as client:
    try:
        client.collections.create(
            "product-embeddings",
            vectors_config=VectorParams(size=1536, distance=Distance.Cosine),
        )
    except CollectionExistsError:
        pass

import time

from actian_vectorai import VectorAIError, get_retry_delay, is_retryable


def search_with_retry(client, collection_name, query_vector, limit=10, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.points.search(
                collection_name,
                vector=query_vector,
                limit=limit,
            )
        except VectorAIError as error:
            if not is_retryable(error) or attempt == max_retries - 1:
                raise
            time.sleep(get_retry_delay(error, attempt=attempt + 1))

from actian_vectorai import VectorAIError

try:
    client.points.search("products", vector=[0.1] * 128, limit=10)
except VectorAIError as error:
    print(f"Code:      {error.code}")
    print(f"Message:   {error.message}")
    print(f"Details:   {error.details}")
    print(f"Operation: {error.operation}")