🚨 Weaviate destination: Add embedding capabilities, overwrite and dedup support, API key auth mode and available on Airbyte Cloud

airbyte-integrations/connectors/destination-weaviate/Dockerfile

@@ -1,16 +1,19 @@
-FROM python:3.9.11-alpine3.15 as base
+FROM python:3.10-slim as base
 
 # build and load all requirements
 FROM base as builder
 WORKDIR /airbyte/integration_code
 
 # upgrade pip to the latest version
-RUN apk --no-cache upgrade \
-    && pip install --upgrade pip \
-    && apk --no-cache add tzdata build-base \
-    && apk add libffi-dev
-
 COPY setup.py ./
+
+RUN pip install --upgrade pip
+
+# This is required because the current connector dependency is not compatible with the CDK version
+# An older CDK version will be used, which depends on pyYAML 5.4, for which we need to pin Cython to <3.0
+# As of today the CDK version that satisfies the main dependency requirements, is 0.1.80 ...
+RUN pip install --prefix=/install "Cython<3.0" "pyyaml~=5.4" --no-build-isolation
+
 # install necessary packages to a temporary folder
 RUN pip install --prefix=/install .
 
@@ -25,7 +28,7 @@ COPY --from=builder /usr/share/zoneinfo/Etc/UTC /etc/localtime
 RUN echo "Etc/UTC" > /etc/timezone
 
 # bash is installed for more convenient debugging.
-RUN apk --no-cache add bash
+RUN apt-get install bash
 
 # copy payload code only
 COPY main.py ./
@@ -34,5 +37,5 @@ COPY destination_weaviate ./destination_weaviate
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.1
-LABEL io.airbyte.name=airbyte/destination-weaviate
+LABEL io.airbyte.version=0.2.0
+LABEL io.airbyte.name=airbyte/destination-weaviate

airbyte-integrations/connectors/destination-weaviate/acceptance-test-config.yml

@@ -0,0 +1,7 @@
+acceptance_tests:
+  spec:
+    tests:
+      - spec_path: integration_tests/spec.json
+        backward_compatibility_tests_config:
+          disable_for_version: "0.2.0"
+connector_image: airbyte/destination-weaviate:dev

airbyte-integrations/connectors/destination-weaviate/acceptance-test-docker.sh

@@ -0,0 +1,2 @@
+#!/usr/bin/env sh
+source "$(git rev-parse --show-toplevel)/airbyte-integrations/bases/connector-acceptance-test/acceptance-test-docker.sh"

airbyte-integrations/connectors/destination-weaviate/destination_weaviate/config.py

@@ -0,0 +1,141 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from typing import List, Literal, Union
+
+import dpath.util
+from airbyte_cdk.destinations.vector_db_based.config import (
+    AzureOpenAIEmbeddingConfigModel,
+    CohereEmbeddingConfigModel,
+    FakeEmbeddingConfigModel,
+    FromFieldEmbeddingConfigModel,
+    OpenAIEmbeddingConfigModel,
+    ProcessingConfigModel,
+)
+from airbyte_cdk.utils.spec_schema_transformations import resolve_refs
+from pydantic import BaseModel, Field
+
+
+class UsernamePasswordAuth(BaseModel):
+    mode: Literal["username_password"] = Field("username_password", const=True)
+    username: str = Field(..., title="Username", description="Username for the Weaviate cluster", order=1)
+    password: str = Field(..., title="Password", description="Password for the Weaviate cluster", airbyte_secret=True, order=2)
+
+    class Config:
+        title = "Username/Password"
+        schema_extra = {"description": "Authenticate using username and password (suitable for self-managed Weaviate clusters)"}
+
+
+class NoAuth(BaseModel):
+    mode: Literal["no_auth"] = Field("no_auth", const=True)
+
+    class Config:
+        title = "No Authentication"
+        schema_extra = {
+            "description": "Do not authenticate (suitable for locally running test clusters, do not use for clusters with public IP addresses)"
+        }
+
+
+class TokenAuth(BaseModel):
+    mode: Literal["token"] = Field("token", const=True)
+    token: str = Field(..., title="API Token", description="API Token for the Weaviate instance", airbyte_secret=True)
+
+    class Config:
+        title = "API Token"
+        schema_extra = {"description": "Authenticate using an API token (suitable for Weaviate Cloud)"}
+
+
+class Header(BaseModel):
+    header_key: str = Field(..., title="Header Key")
+    value: str = Field(..., title="Header Value", airbyte_secret=True)
+
+
+class WeaviateIndexingConfigModel(BaseModel):
+    host: str = Field(
+        ...,
+        title="Public Endpoint",
+        order=1,
+        description="The public endpoint of the Weaviate cluster.",
+        examples=["https://my-cluster.weaviate.network"],
+    )
+    auth: Union[TokenAuth, UsernamePasswordAuth, NoAuth] = Field(
+        ..., title="Authentication", description="Authentication method", discriminator="mode", type="object", order=2
+    )
+    batch_size: int = Field(title="Batch Size", description="The number of records to send to Weaviate in each batch", default=128)
+    text_field: str = Field(title="Text Field", description="The field in the object that contains the embedded text", default="text")
+    default_vectorizer: str = Field(
+        title="Default Vectorizer",
+        description="The vectorizer to use if new classes need to be created",
+        default="none",
+        enum=[
+            "none",
+            "text2vec-cohere",
+            "text2vec-huggingface",
+            "text2vec-openai",
+            "text2vec-palm",
+            "text2vec-contextionary",
+            "text2vec-transformers",
+            "text2vec-gpt4all",
+        ],
+    )
+    additional_headers: List[Header] = Field(
+        title="Additional headers",
+        description="Additional HTTP headers to send with every request.",
+        default=[],
+        examples=[{"header_key": "X-OpenAI-Api-Key", "value": "my-openai-api-key"}],
+    )
+
+    class Config:
+        title = "Indexing"
+        schema_extra = {
+            "group": "indexing",
+            "description": "Indexing configuration",
+        }
+
+
+class NoEmbeddingConfigModel(BaseModel):
+    mode: Literal["no_embedding"] = Field("no_embedding", const=True)
+
+    class Config:
+        title = "No external embedding"
+        schema_extra = {
+            "description": "Do not calculate and pass embeddings to Weaviate. Suitable for clusters with configured vectorizers to calculate embeddings within Weaviate or for classes that should only support regular text search."
+        }
+
+
+class ConfigModel(BaseModel):
+    processing: ProcessingConfigModel
+    embedding: Union[
+        NoEmbeddingConfigModel,
+        AzureOpenAIEmbeddingConfigModel,
+        OpenAIEmbeddingConfigModel,
+        CohereEmbeddingConfigModel,
+        FromFieldEmbeddingConfigModel,
+        FakeEmbeddingConfigModel,
+    ] = Field(..., title="Embedding", description="Embedding configuration", discriminator="mode", group="embedding", type="object")
+    indexing: WeaviateIndexingConfigModel
+
+    class Config:
+        title = "Weaviate Destination Config"
+        schema_extra = {
+            "groups": [
+                {"id": "processing", "title": "Processing"},
+                {"id": "embedding", "title": "Embedding"},
+                {"id": "indexing", "title": "Indexing"},
+            ]
+        }
+
+    @staticmethod
+    def remove_discriminator(schema: dict) -> None:
+        """pydantic adds "discriminator" to the schema for oneOfs, which is not treated right by the platform as we inline all references"""
+        dpath.util.delete(schema, "properties/*/discriminator")
+        dpath.util.delete(schema, "properties/**/discriminator")
+
+    @classmethod
+    def schema(cls):
+        """we're overriding the schema classmethod to enable some post-processing"""
+        schema = super().schema()
+        schema = resolve_refs(schema)
+        cls.remove_discriminator(schema)
+        return schema