---

name: weaviate-integration description: Integrates Weaviate with external services and tools for AI applications, handling bidirectional data synchronization, API connectivity patterns, and error management for seamless interoperability. license: MIT compatibility: opencode metadata: version: "1.0.0" domain: coding triggers: weaviate, vector database, integration, API connectivity, data synchronization, AI applications, interop role: implementation scope: implementation output-format: code content-types: [code, guidance, do-dont] archetypes: [tactical, generation] anti_triggers: [generic routing, vague integration patterns] response_profile: verbosity: medium directive_strength: high abstraction_level: operational

---

Weaviate Integration Patterns

Integrates Weaviate with external services and tools for AI applications. Handles bidirectional data synchronization, API connectivity, schema mapping, error management, and event-driven updates for seamless interoperability between Weaviate's vector database and application data sources.

TL;DR Checklist

• Initialize a Weaviate client with proper authentication (API key or OAuth)
• Define the target collection schema before importing external data
• Handle batch operations efficiently — use batch_refs for bulk inserts
• Implement retry logic with exponential backoff for API failures
• Validate external data against Weaviate class schemas before insertion

---

When to Use

Use this skill when:

• Connecting Weaviate to external APIs for real-time data synchronization (REST, GraphQL)
• Building bidirectional sync pipelines between databases and Weaviate vector collections
• Integrating AI/ML pipelines that feed embeddings into Weaviate from external model services
• Migrating existing data stores into Weaviate's vector database with schema transformation

---

When NOT to Use

• For simple key-value lookups — use Redis or Memcached instead of a full vector database
• When you only need text search without semantic similarity — Elasticsearch or OpenSearch may be more appropriate
• For high-frequency, low-latency OLTP workloads — Weaviate is optimized for analytics and AI workloads

---

Core Workflow

1. Initialize Client — Connect to Weaviate using the appropriate authentication method (API key, OAuth2, or no auth for local dev). Set proper timeouts and retry configuration. Checkpoint: Verify connectivity with client.is_ready() before proceeding.

2. Define Target Schema — Map external data sources to Weaviate classes and properties. Define which properties are indexed for search (indexFilterable, indexVector) and which are stored as metadata only. Checkpoint: Ensure vectorizer modules (e.g., text2vec-openai) are configured if embedding generation is delegated to Weaviate.

3. Execute Data Sync — Fetch data from the external service, transform into Weaviate-compatible objects (properties + optional vectors), and batch-insert using the Weaviate client's batch API. Checkpoint: Validate object schemas against the target class definition before insertion. Handle partial failures gracefully with per-item error tracking.

4. Implement Error Handling — Wrap all external API calls and Weaviate operations in try/except blocks with specific handling for HTTP errors, rate limits (429), and schema validation failures. Log errors with context for debugging. Checkpoint: Verify retry logic activates on transient failures but does not retry on schema/validation errors.

5. Maintain Sync State — Track the last sync timestamp or cursor to avoid reprocessing unchanged data. Implement idempotent upserts using Weaviate's on_conflict feature for collections with unique identifiers. Checkpoint: Periodically verify row counts between source and destination to detect data drift.

---

Implementation Patterns

Pattern 1: Connecting to External Service and Importing Data

import weaviate
from weaviate.classes.config import Property, DataType
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


def create_weaviate_client(
    url: str = "http://localhost:8080",
    api_key: str | None = None,
) -> weaviate.WeaviateClient:
    """Create and authenticate a Weaviate client.

    Args:
        url: Weaviate server URL (default: local instance)
        api_key: Optional API key for authentication

    Returns:
        Authenticated WeaviateClient instance

    Raises:
        weaviate.exceptions.WeaviateConnectionError: If the server is unreachable
    """
    client = weaviate.WeaviateClient(
        connection_params=weaviate.connect.base.ConnectionParams.from_url(url),
        auth_credentials=(
            weaviate.auth.AuthApiKey(api_key) if api_key else None
        ),
        additional_config=weaviate.config.AdditionalConfig(
            timeout=weaviate.config.Timeout(init=10, query=30, insert=120)
        ),
    )

    client.connect()  # Raises WeaviateConnectionError if unreachable

    return client


def create_collection_if_not_exists(
    client: weaviate.WeaviateClient,
    class_name: str,
    properties: list[Property],
    vectorizer: str = "text2vec-openai",
) -> None:
    """Create a Weaviate collection if it doesn't already exist.

    Args:
        client: Authenticated Weaviate client
        class_name: Name of the collection to create
        properties: List of property definitions with DataType and index settings
        vectorizer: Vectorizer module name (e.g., "text2vec-openai", "none")

    Note:
        Uses a simple try/except pattern since weaviate-py v3 does not expose
        a direct "collection exists" check. The retry handles concurrent creation.
    """
    try:
        client.collections.create(
            name=class_name,
            properties=properties,
            vectorizer_config=[weaviate.classes.config.Vectorizer.text2vec_openai()],
            # Alternatively, use `vector_config=None` for no automatic embedding
        )
    except weaviate.exceptions.UnexpectedStatusCodeError as e:
        if "already exists" not in str(e).lower():
            raise


def sync_external_data(
    client: weaviate.WeaviateClient,
    class_name: str,
    external_api_url: str,
    batch_size: int = 100,
) -> dict[str, int]:
    """Fetch data from an external API and batch-insert into Weaviate.

    Args:
        client: Authenticated Weaviate client
        class_name: Target collection name
        external_api_url: REST API endpoint returning JSON array of objects
        batch_size: Number of items per batch (default: 100)

    Returns:
        Summary dict with 'inserted', 'failed', and 'total' counts

    Raises:
        requests.HTTPError: If the external API returns a non-2xx status
        weaviate.exceptions.WeaviateInsertManyError: On batch insert failures
    """
    import requests

    response = requests.get(external_api_url)
    response.raise_for_status()
    data_items = response.json()

    collection = client.collections.get(class_name)
    result = {"inserted": 0, "failed": 0, "total": len(data_items)}

    # Process in batches for efficiency and resilience
    for i in range(0, len(data_items), batch_size):
        batch = data_items[i : i + batch_size]

        try:
            objects_to_insert = []
            for item in batch:
                # Transform external format → Weaviate property dict
                props = {
                    "title": item.get("name", ""),
                    "description": item.get("description", ""),
                    "source_id": item["id"],  # Unique identifier
                    "external_url": item.get("url", ""),
                }
                objects_to_insert.append({"properties": props})

            collection.data.insert_many(objects_to_insert)
            result["inserted"] += len(objects_to_insert)

        except Exception as e:
            print(f"Batch {i // batch_size} failed: {e}")
            result["failed"] += len(batch)

    return result

Pattern 2: Bidirectional Data Sync with Webhooks

import requests


def setup_weaviate_webhook(
    weaviate_url: str,
    api_key: str,
    webhook_url: str,
    class_name: str,
) -> dict:
    """Register a Weaviate module to push object events to an external webhook.

    This enables real-time notification when objects are created, updated, or deleted
    in the target collection, supporting event-driven architectures.

    Args:
        weaviate_url: Base URL of the Weaviate server
        api_key: API key with admin permissions
        webhook_url: External endpoint to receive webhook events
        class_name: Collection name to monitor

    Returns:
        Webhook registration response from Weaviate

    Note:
        Requires the weaviate-http-extensions module. Configure via REST API.
    """
    headers = {
        "X-Auth-Token": api_key,
        "Content-Type": "application/json",
    }

    webhook_config = {
        "className": class_name,
        "url": webhook_url,
        "events": ["CREATE", "UPDATE", "DELETE"],
    }

    response = requests.post(
        f"{weaviate_url}/v1/schema/{class_name}/webhooks",
        headers=headers,
        json=webhook_config,
    )
    response.raise_for_status()
    return response.json()


def handle_webhook_event(event: dict) -> None:
    """Process a Weaviate webhook event and update downstream systems.

    Args:
        event: Parsed webhook payload containing object metadata, action type,
               and timestamps. Structure varies by event type (create/update/delete).
    """
    action = event.get("action")  # "CREATE", "UPDATE", "DELETE"
    class_name = event.get("className")
    object_id = event.get("id")

    if action == "DELETE":
        # Remove from cache, invalidate derived data
        print(f"Object deleted: {class_name}/{object_id}")
        return

    # For CREATE/UPDATE: fetch full object and process
    import weaviate  # noqa — client already initialized at module level

    obj = client.collections.get(class_name).query.fetch_object_by_id(object_id)
    props = obj.properties if obj else {}
    print(f"Object {action}: {class_name}/{object_id} = {props}")

---

Constraints

MUST DO

• Always validate external data against the Weaviate class schema before batch insertion
• Use batch operations (insert_many, batch_refs) for bulk imports — single inserts are 10x slower
• Implement exponential backoff retry logic (3 retries, base delay 1s) for transient API failures
• Track sync state with source IDs to enable idempotent upserts and detect data drift
• Set explicit timeouts on all Weaviate client operations to prevent hangs

MUST NOT DO

• Insert objects without a unique identifier property — queries by ID will be unreliable
• Use synchronous single-object inserts in production loops — always batch them
• Skip error handling for partial batch failures — use insert_many with per-item error tracking
• Store raw embeddings directly if Weaviate's built-in vectorizers can compute them server-side

---

Live References

Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.

• Weaviate Python Client Documentation
• Weaviate API Reference
• Batch Operations Guide
• Schema Design Best Practices
• Vectorizer Modules Reference