🎉 New Destination deepset

airbyte-integrations/connectors/destination-deepset/README.md

@@ -0,0 +1,103 @@
+# Deepset Destination
+
+This is the repository for the Deepset destination connector, written in Python.
+For information about how to use this connector within Airbyte, see [the documentation](https://docs.airbyte.com/integrations/destinations/deepset).
+
+## Local development
+
+### Prerequisites
+
+* Python (`^3.9`)
+* Poetry (`^1.7`) - installation instructions [here](https://python-poetry.org/docs/#installation)
+
+
+
+### Installing the connector
+
+From this connector directory, run:
+```bash
+poetry install --with dev
+```
+
+
+#### Create credentials
+
+**If you are a community contributor**, follow the instructions in the [documentation](https://docs.airbyte.com/integrations/destinations/deepset)
+to generate the necessary credentials. Then create a file `secrets/config.json` conforming to the `destination_deepset/spec.json` file.
+Note that the `secrets` directory is gitignored by default, so there is no danger of accidentally checking in sensitive information.
+See `integration_tests/sample_config.json` for a sample config file.
+
+**If you are an Airbyte core member**, copy the credentials in Lastpass under the secret name `destination deepset test creds`
+and place them into `secrets/config.json`.
+
+### Locally running the connector
+```
+poetry run destination-deepset spec
+poetry run destination-deepset check --config secrets/config.json
+poetry run destination-deepset write --config secrets/config.json --catalog sample_files/configured_catalog.json
+```
+
+### Running tests
+
+To run tests locally, from the connector directory run:
+
+```
+poetry run pytest tests
+```
+
+### Building the docker image
+
+1. Install [`airbyte-ci`](https://github.com/airbytehq/airbyte/blob/master/airbyte-ci/connectors/pipelines/README.md)
+2. Run the following command to build the docker image:
+```bash
+airbyte-ci connectors --name=destination-deepset build
+```
+
+An image will be available on your host with the tag `airbyte/destination-deepset:dev`.
+
+### Running as a docker container
+
+Then run any of the connector commands as follows:
+```
+docker run --rm airbyte/destination-deepset:dev spec
+docker run --rm -v $(pwd)/secrets:/secrets airbyte/destination-deepset:dev check --config /secrets/config.json
+docker run --rm -v $(pwd)/secrets:/secrets -v $(pwd)/integration_tests:/integration_tests airbyte/destination-deepset:dev write --config /secrets/config.json --catalog /integration_tests/configured_catalog.json
+```
+
+### Running our CI test suite
+
+You can run our full test suite locally using [`airbyte-ci`](https://github.com/airbytehq/airbyte/blob/master/airbyte-ci/connectors/pipelines/README.md):
+
+```bash
+airbyte-ci connectors --name=destination-deepset test
+```
+
+### Customizing acceptance Tests
+
+Customize `acceptance-test-config.yml` file to configure acceptance tests. See [Connector Acceptance Tests](https://docs.airbyte.com/connector-development/testing-connectors/connector-acceptance-tests-reference) for more information.
+If your connector requires to create or destroy resources for use during acceptance tests create fixtures for it and place them inside integration_tests/acceptance.py.
+
+### Dependency Management
+
+All of your dependencies should be managed via Poetry.
+To add a new dependency, run:
+
+```bash
+poetry add <package-name>
+```
+
+Please commit the changes to `pyproject.toml` and `poetry.lock` files.
+
+## Publishing a new version of the connector
+
+You've checked out the repo, implemented a million dollar feature, and you're ready to share your changes with the world. Now what?
+1. Make sure your changes are passing our test suite: `airbyte-ci connectors --name=destination-deepset test`
+2. Bump the connector version (please follow [semantic versioning for connectors](https://docs.airbyte.com/contributing-to-airbyte/resources/pull-requests-handbook/#semantic-versioning-for-connectors)):
+    - bump the `dockerImageTag` value in in `metadata.yaml`
+    - bump the `version` value in `pyproject.toml`
+3. Make sure the `metadata.yaml` content is up to date.
+4. Make sure the connector documentation and its changelog is up to date (`docs/integrations/destinations/deepset.md`).
+5. Create a Pull Request: use [our PR naming conventions](https://docs.airbyte.com/contributing-to-airbyte/resources/pull-requests-handbook/#pull-request-title-convention).
+6. Pat yourself on the back for being an awesome contributor.
+7. Someone from Airbyte will take a look at your PR and iterate with you to merge it into master.
+8. Once your PR is merged, the new version of the connector will be automatically published to Docker Hub and our connector registry.

airbyte-integrations/connectors/destination-deepset/destination_deepset/__init__.py

@@ -0,0 +1,9 @@
+#
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+#
+from __future__ import annotations
+
+from .destination import DestinationDeepset
+
+
+__all__ = ["DestinationDeepset"]

airbyte-integrations/connectors/destination-deepset/destination_deepset/api.py

@@ -0,0 +1,125 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+
+from __future__ import annotations
+
+from uuid import UUID
+
+import httpx
+from destination_deepset.models import SUPPORTED_FILE_EXTENSIONS, DeepsetCloudConfig, DeepsetCloudFile
+from httpx import HTTPError, HTTPStatusError
+from tenacity import Retrying, retry_if_exception_type, stop_after_attempt, wait_random_exponential
+
+
+class APIError(RuntimeError):
+    """Raised when any error occurs while using the API."""
+
+
+class ConfigurationError(ValueError, APIError):
+    """Raised when the configuration is missing or incorrect."""
+
+    def __str__(self) -> str:
+        return "Configuration is missing, cannot create an HTTP client."
+
+
+class FileTypeError(APIError):
+    """Raised when the file's extension does not match one of the supported file types."""
+
+    def __str__(self) -> str:
+        return f"File type not supported. Supported file extensions: {SUPPORTED_FILE_EXTENSIONS}"
+
+
+class FileUploadError(APIError):
+    """Raised when the server is unable to successfully upload the file."""
+
+    def __str__(self) -> str:
+        return "File upload failed."
+
+
+class DeepsetCloudApi:
+    def __init__(self, config: DeepsetCloudConfig) -> None:
+        self.config = config
+        self._client: httpx.Client | None = None
+
+        # retry settings in seconds
+        self.max = 60
+        self.multiplier = 0.5
+
+    @property
+    def client(self) -> httpx.Client:
+        if not self.config:
+            raise ConfigurationError
+
+        if self._client is None:
+            self._client = httpx.Client(
+                base_url=self.config.base_url,
+                headers={
+                    "Accept": "application/json",
+                    "Authorization": f"Bearer {self.config.api_key}",
+                    "X-Client-Source": "airbyte-destination-deepset",
+                },
+                follow_redirects=True,
+            )
+
+        return self._client
+
+    def health_check(self) -> None:
+        """Check the health of deepset cloud API
+
+        Raises:
+            APIError: Raised when an error is encountered.
+        """
+        try:
+            for attempt in Retrying(
+                retry=retry_if_exception_type(HTTPError),
+                stop=stop_after_attempt(self.config.retries),
+                wait=wait_random_exponential(multiplier=self.multiplier, max=self.max),
+                reraise=True,
+            ):
+                with attempt:
+                    response = self.client.get(f"/api/v1/workspaces/{self.config.workspace}/files")
+                    response.raise_for_status()
+        except Exception as ex:
+            raise APIError from ex
+
+    def upload(self, file: DeepsetCloudFile) -> UUID:
+        """Upload file to deepset Cloud.
+
+        Args:
+            file (DeepsetCloudFile): The file to upload
+
+        Raises:
+            APIError: Raised whenever the file upload fails
+
+        Returns:
+            UUID: The unique identifier of the uploaded file
+        """
+        if file.extension not in SUPPORTED_FILE_EXTENSIONS:
+            raise FileTypeError
+
+        try:
+            for attempt in Retrying(
+                retry=retry_if_exception_type(HTTPError),
+                stop=stop_after_attempt(self.config.retries),
+                wait=wait_random_exponential(multiplier=self.multiplier, max=self.max),
+                reraise=True,
+            ):
+                with attempt:
+                    response = self.client.post(
+                        f"/api/v1/workspaces/{self.config.workspace}/files",
+                        files={"file": (file.name, file.content)},
+                        data={"meta": file.meta_as_string},
+                        params={"write_mode": "OVERWRITE"},
+                    )
+                    response.raise_for_status()
+
+            if file_id := response.json().get("file_id"):
+                return UUID(file_id)
+
+        except HTTPStatusError as ex:
+            status_code, response_text = ex.response.status_code, ex.response.text
+            message = f"File upload failed: {status_code = }, {response_text = }."
+            raise FileUploadError(message) from ex
+        except Exception as ex:
+            raise FileUploadError from ex
+
+        raise FileUploadError

airbyte-integrations/connectors/destination-deepset/destination_deepset/destination.py

@@ -0,0 +1,89 @@
+#
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+#
+from __future__ import annotations
+
+import logging
+from collections.abc import Iterable, Mapping
+from typing import Any
+
+from airbyte_cdk.destinations import Destination
+from airbyte_cdk.models import AirbyteConnectionStatus, AirbyteMessage, ConfiguredAirbyteCatalog, Status, Type
+from destination_deepset import util
+from destination_deepset.models import DeepsetCloudFile, Filetypes
+from destination_deepset.writer import DeepsetCloudFileWriter
+
+logger = logging.getLogger("airbyte")
+
+
+class DestinationDeepset(Destination):
+    def write(
+        self,
+        config: Mapping[str, Any],
+        configured_catalog: ConfiguredAirbyteCatalog,
+        input_messages: Iterable[AirbyteMessage],
+    ) -> Iterable[AirbyteMessage]:
+        """Reads the input stream of messages, config, and catalog to write data to the destination.
+
+        This method returns an iterable (typically a generator of AirbyteMessages via yield) containing state messages
+        received in the input message stream. Outputting a state message means that every AirbyteRecordMessage which
+        came before it has been successfully persisted to the destination. This is used to ensure fault tolerance in the
+        case that a sync fails before fully completing, then the source is given the last state message output from this
+        method as the starting point of the next sync.
+
+        Args:
+            config (Mapping[str, Any]): dict of JSON configuration matching the configuration declared in spec.json
+            configured_catalog (ConfiguredAirbyteCatalog): The Configured Catalog describing the schema of the data
+                being received and how it should be persisted in the destination
+            input_messages (Iterable[AirbyteMessage]): The stream of input messages received from the source
+
+        Returns:
+            Iterable[AirbyteMessage]: Iterable of AirbyteStateMessages wrapped in AirbyteMessage structs
+        """
+        writer = DeepsetCloudFileWriter.factory(config)
+
+        streams: set[str] = set()
+        for catalog_stream in configured_catalog.streams:
+            if util.get(stream := catalog_stream.stream, "format.filetype") == Filetypes.DOCUMENT:
+                streams.add(stream.name)
+            else:
+                logger.warning(f"Only sources that produce document file type formatted records are supported! Got {stream = !s}")
+
+        for message in input_messages:
+            match message.type:
+                case Type.STATE:
+                    yield message
+                case Type.RECORD:
+                    if (stream_name := message.record.stream) not in streams:
+                        logger.debug(f"Stream {stream_name} was not present in configured streams, skipping")
+                        continue
+
+                    try:
+                        file = DeepsetCloudFile.from_record(message.record)
+                    except ValueError as ex:
+                        yield util.get_trace_message("Failed to parse data into deepset cloud file instance.", exception=ex)
+                    else:
+                        yield writer.write(file=file)
+                case _:
+                    continue
+
+    def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteConnectionStatus:
+        """Tests if the input configuration can be used to successfully connect to the destination with the needed
+        permissions e.g: if a provided API token or password can be used to connect and write to the destination.
+
+        Args:
+            logger (logging.Logger): Logging object to display debug/info/error to the logs (logs will not be accessible
+                via airbyte UI if they are not passed to this logger)
+            config (Mapping[str, Any]): Json object containing the configuration of this destination, content of this
+                json is as specified in the properties of the spec.json file
+
+        Returns:
+            AirbyteConnectionStatus: AirbyteConnectionStatus indicating a Success or Failure
+        """
+        try:
+            writer = DeepsetCloudFileWriter.factory(config)
+            writer.client.health_check()
+        except Exception as ex:
+            return AirbyteConnectionStatus(status=Status.FAILED, message=f"Failed to connect to deepset cloud, reason: {ex!s}")
+        else:
+            return AirbyteConnectionStatus(status=Status.SUCCEEDED)