Weaviate Cookbooks

import weaviate
from weaviate.classes.init import Auth

# Use the official helper function for Weaviate Cloud
client = weaviate.use_async_with_weaviate_cloud(
    cluster_url="your-cluster.weaviate.cloud",  # Accepts hostname with or without https://
    auth_credentials=Auth.api_key("your-api-key"),
    headers={  # Note: parameter is "headers" not "additional_headers"
        "X-OpenAI-Api-Key": "sk-...",
        "X-Anthropic-Api-Key": "sk-ant-...",
    }
)

await client.connect()  # Required! Async helpers don't auto-connect

# For local instances
client = weaviate.use_async_with_local()

# For custom endpoints
client = weaviate.use_async_with_custom(
    http_host="localhost",
    http_port=8080,
    http_secure=False,
    grpc_host="localhost",
    grpc_port=50051,
    grpc_secure=False,
)

await client.connect()

from weaviate.classes.init import Auth

# API Key (most common for Weaviate Cloud)
auth = Auth.api_key("your-api-key")

# Bearer Token (with optional refresh token)
auth = Auth.bearer_token("access-token", refresh_token="refresh-token")

# Client Credentials (OIDC)
auth = Auth.client_credentials(client_secret="secret")

# Client Password (OIDC Resource Owner Password flow)
auth = Auth.client_password(username="user", password="pass")

# Usage
client = weaviate.use_async_with_weaviate_cloud(
    cluster_url="your-cluster.weaviate.cloud",
    auth_credentials=auth,
)

# ❌ Wrong - client not connected
client = weaviate.use_async_with_weaviate_cloud(...)
collections = await client.collections.list_all()  # Will fail!

# ✅ Correct - explicit connect/close
client = weaviate.use_async_with_weaviate_cloud(...)
await client.connect()
collections = await client.collections.list_all()
await client.close()

# Collection retrieval is SYNC (no await)
collection = client.collections.get("MyCollection")

# Operations on collections are ASYNC (need await)
config = await collection.config.get()
results = await collection.query.fetch_objects()
count = await collection.aggregate.over_all()

# ✅ For bulk imports, prefer sync client
import weaviate

with weaviate.connect_to_weaviate_cloud(...) as client:
    collection = client.collections.get("MyCollection")

    # Batch insert handles concurrency automatically
    with collection.batch.dynamic() as batch:
        for item in large_dataset:
            batch.add_object(properties=item)

from contextlib import asynccontextmanager
from typing import AsyncGenerator

@asynccontextmanager
async def get_weaviate_client(
    cluster_url: str,
    api_key: str,
    provider_headers: dict[str, str] | None = None,
) -> AsyncGenerator[weaviate.WeaviateAsyncClient, None]:
    """Connect to Weaviate Cloud with automatic cleanup."""
    # Remove scheme if present
    hostname = cluster_url.replace("https://", "").replace("http://", "")

    client = weaviate.use_async_with_weaviate_cloud(
        cluster_url=hostname,
        auth_credentials=Auth.api_key(api_key),
        headers=provider_headers,
    )

    try:
        await client.connect()
        yield client
    finally:
        await client.close()

# Usage
async def example():
    async with get_weaviate_client(
        cluster_url="your-cluster.weaviate.cloud",
        api_key="your-key",
    ) as client:
        collections = await client.collections.list_all()

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: connect to Weaviate
    app.state.weaviate = weaviate.use_async_with_weaviate_cloud(
        cluster_url="your-cluster.weaviate.cloud",
        auth_credentials=Auth.api_key("your-key"),
    )
    await app.state.weaviate.connect()

    yield

    # Shutdown: close connection
    await app.state.weaviate.close()

app = FastAPI(lifespan=lifespan)

@app.get("/collections")
async def list_collections():
    collections = await app.state.weaviate.collections.list_all()
    return {"collections": list(collections.keys())}

# ❌ Wrong - WeaviateAsyncClient() constructor uses different param
client = weaviate.use_async_with_weaviate_cloud(
    additional_headers={...}  # Wrong parameter name!
)

# ✅ Correct - use "headers" not "additional_headers"
client = weaviate.use_async_with_weaviate_cloud(
    headers={...}
)

# ✅ Both accepted
client = weaviate.use_async_with_weaviate_cloud(
    cluster_url="https://cluster.weaviate.cloud"  # With scheme
)

client = weaviate.use_async_with_weaviate_cloud(
    cluster_url="cluster.weaviate.cloud"  # Without scheme
)

# ❌ Wrong - sync client (cannot use await)
client = weaviate.connect_to_weaviate_cloud(...)
await client.connect()  # TypeError!

# ✅ Correct - async client
client = weaviate.use_async_with_weaviate_cloud(...)
await client.connect()

# ❌ Wrong - manual port config causes conflicts with Weaviate Cloud
client = WeaviateAsyncClient(
    connection_params=ConnectionParams.from_url(
        url="https://cluster.weaviate.cloud",
        grpc_port=443,  # Conflict!
    )
)

# ✅ Correct - use helper function (handles ports automatically)
client = weaviate.use_async_with_weaviate_cloud(
    cluster_url="cluster.weaviate.cloud"
)

@asynccontextmanager
async def get_multi_cluster_clients(
    clusters: dict[str, dict[str, str]]
) -> AsyncGenerator[dict[str, weaviate.WeaviateAsyncClient], None]:
    """Connect to multiple Weaviate clusters.

    Args:
        clusters: Dict of {cluster_id: {"url": "...", "api_key": "..."}}
    """
    clients = {}

    try:
        # Connect to all clusters
        for cluster_id, config in clusters.items():
            client = weaviate.use_async_with_weaviate_cloud(
                cluster_url=config["url"],
                auth_credentials=Auth.api_key(config["api_key"]),
            )
            await client.connect()
            clients[cluster_id] = client

        yield clients

    finally:
        # Close all connections
        for client in clients.values():
            await client.close()

# Usage
async def example():
    clusters = {
        "prod": {"url": "prod.weaviate.cloud", "api_key": "key1"},
        "dev": {"url": "dev.weaviate.cloud", "api_key": "key2"},
    }

    async with get_multi_cluster_clients(clusters) as clients:
        prod_collections = await clients["prod"].collections.list_all()
        dev_collections = await clients["dev"].collections.list_all()

import os

# Read from environment
cluster_url = os.environ["WEAVIATE_URL"]
api_key = os.environ["WEAVIATE_API_KEY"]

# Build provider headers
provider_headers = {}
if openai_key := os.getenv("OPENAI_API_KEY"):
    provider_headers["X-OpenAI-Api-Key"] = openai_key
if anthropic_key := os.getenv("ANTHROPIC_API_KEY"):
    provider_headers["X-Anthropic-Api-Key"] = anthropic_key

client = weaviate.use_async_with_weaviate_cloud(
    cluster_url=cluster_url,
    auth_credentials=Auth.api_key(api_key),
    headers=provider_headers or None,
)

import pytest

@pytest.mark.asyncio
async def test_weaviate_connection():
    async with get_weaviate_client(
        cluster_url="test-cluster.weaviate.cloud",
        api_key="test-key",
    ) as client:
        collections = await client.collections.list_all()
        assert isinstance(collections, dict)