Source DuckDB: ✨ Add MotherDuck support 🦆🦆

.devcontainer/destination-duckdb/devcontainer.json

@@ -0,0 +1,76 @@
+// For format details, see https://aka.ms/devcontainer.json. For config options, see the
+// README at: https://github.com/devcontainers/templates/tree/main/src/python
+{
+  "name": "DuckDB Destination Connector DevContainer (Python)",
+
+  // Or use a Dockerfile or Docker Compose file. More info: https://containers.dev/guide/dockerfile
+  "image": "mcr.microsoft.com/devcontainers/python:0-3.9",
+
+  // Features to add to the dev container. More info: https://containers.dev/features.
+  "features": {
+    "ghcr.io/devcontainers-contrib/features/poetry:2": {},
+    "ghcr.io/devcontainers/features/docker-in-docker": {}
+  },
+  "overrideFeatureInstallOrder": [
+    // Deterministic order maximizes cache reuse
+    "ghcr.io/devcontainers-contrib/features/poetry",
+    "ghcr.io/devcontainers/features/docker-in-docker"
+  ],
+
+  "workspaceFolder": "/workspaces/airbyte/airbyte-integrations/connectors/destination-duckdb",
+
+  // Configure tool-specific properties.
+  "customizations": {
+    "vscode": {
+      "extensions": [
+				// Python extensions:
+				"charliermarsh.ruff",
+				"matangover.mypy",
+				"ms-python.black",
+				"ms-python.python",
+				"ms-python.vscode-pylance",
+
+				// Toml support
+				"tamasfe.even-better-toml",
+
+				// Yaml and JSON Schema support:
+				"redhat.vscode-yaml",
+
+				// Contributing:
+				"GitHub.vscode-pull-request-github"
+      ],
+      "settings": {
+        "extensions.ignoreRecommendations": true,
+        "git.openRepositoryInParentFolders": "always",
+        "python.defaultInterpreterPath": ".venv/bin/python",
+        "python.interpreter.infoVisibility": "always",
+        "python.terminal.activateEnvironment": true,
+        "python.testing.pytestEnabled": true,
+        "python.testing.cwd": "/workspaces/airbyte/airbyte-integrations/connectors/destination-duckdb",
+        "python.testing.pytestArgs": [
+          "--rootdir=/workspaces/airbyte/airbyte-integrations/connectors/destination-duckdb",
+          "."
+        ]
+      }
+    }
+  },
+  "containerEnv": {
+    "POETRY_VIRTUALENVS_IN_PROJECT": "true"
+  },
+
+  // Mark the root directory as 'safe' for git.
+  "initializeCommand": "git config --add safe.directory /workspaces/airbyte",
+
+  // Use 'postCreateCommand' to run commands after the container is created.
+  // Post-create tasks:
+  // 1. Create a symlink directory.
+  // 2. Create symlinks for the devcontainer.json and docs markdown file.
+  // 3. Install the Python/Poetry dependencies.
+  "postCreateCommand": "mkdir -p ./.symlinks && echo '*' > ./.symlinks/.gitignore && ln -sf /workspaces/airbyte/.devcontainer/destination-duckdb/devcontainer.json ./.symlinks/devcontainer.json && ln -sf /workspaces/airbyte/docs/integrations/destinations/duckdb.md ./.symlinks/duckdb-docs.md && poetry install"
+
+  // Use 'forwardPorts' to make a list of ports inside the container available locally.
+  // "forwardPorts": [],
+
+  // Uncomment to connect as root instead. More info: https://aka.ms/dev-containers-non-root.
+  // "remoteUser": "root"
+}

airbyte-ci/connectors/connector_ops/connector_ops/utils.py

@@ -250,7 +250,8 @@ def metadata(self) -> Optional[dict]:
     def language(self) -> ConnectorLanguage:
         if Path(self.code_directory / self.technical_name.replace("-", "_") / "manifest.yaml").is_file():
             return ConnectorLanguage.LOW_CODE
-        if Path(self.code_directory / "setup.py").is_file():
+        if Path(self.code_directory / "setup.py").is_file() or Path(self.code_directory / "pyproject.toml").is_file():
+            return ConnectorLanguage.PYTHON
             return ConnectorLanguage.PYTHON
         try:
             with open(self.code_directory / "Dockerfile") as dockerfile:

airbyte-ci/connectors/connector_ops/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "connector_ops"
-version = "0.2.2"
+version = "0.2.3"
 description = "Packaged maintained by the connector operations team to perform CI for connectors"
 authors = ["Airbyte <contact@airbyte.io>"]
 

airbyte-ci/connectors/pipelines/pipelines/actions/environments.py

@@ -283,6 +283,35 @@ async def find_local_dependencies_in_pyproject_toml(
     return local_dependency_paths
 
 
+def _install_python_dependencies_from_setup_py(
+    container: Container,
+    additional_dependency_groups: Optional[List] = None,
+) -> Container:
+    install_connector_package_cmd = ["python", "-m", "pip", "install", "."]
+    container = container.with_exec(install_connector_package_cmd)
+
+    if additional_dependency_groups:
+        # e.g. .[dev,tests]
+        group_string = f".[{','.join(additional_dependency_groups)}]"
+        group_install_cmd = ["python", "-m", "pip", "install", group_string]
+
+        container = container.with_exec(group_install_cmd)
+
+    return container
+
+
+def _install_python_dependencies_from_requirements_txt(container: Container) -> Container:
+    install_requirements_cmd = ["python", "-m", "pip", "install", "-r", "requirements.txt"]
+    return container.with_exec(install_requirements_cmd)
+
+
+def _install_python_dependencies_from_poetry(container: Container) -> Container:
+    pip_install_poetry_cmd = ["python", "-m", "pip", "install", "poetry"]
+    poetry_disable_virtual_env_cmd = ["poetry", "config", "virtualenvs.create", "false"]
+    poetry_install_no_venv_cmd = ["poetry", "install", "--no-root", "--no-dev"]
+    return container.with_exec(pip_install_poetry_cmd).with_exec(poetry_disable_virtual_env_cmd).with_exec(poetry_install_no_venv_cmd)
+
+
 async def with_installed_python_package(
     context: PipelineContext,
     python_environment: Container,
@@ -302,29 +331,25 @@ async def with_installed_python_package(
     Returns:
         Container: A python environment container with the python package installed.
     """
-    install_requirements_cmd = ["python", "-m", "pip", "install", "-r", "requirements.txt"]
-    install_connector_package_cmd = ["python", "-m", "pip", "install", "."]
-
     container = with_python_package(context, python_environment, package_source_code_path, exclude=exclude)
 
     local_dependencies = await find_local_python_dependencies(context, package_source_code_path)
 
     for dependency_directory in local_dependencies:
         container = container.with_mounted_directory("/" + dependency_directory, context.get_repo_dir(dependency_directory))
 
-    has_setup_py, has_requirements_txt = await check_path_in_workdir(container, "setup.py"), await check_path_in_workdir(
-        container, "requirements.txt"
-    )
+    has_setup_py = await check_path_in_workdir(container, "setup.py")
+    has_requirements_txt = await check_path_in_workdir(container, "requirements.txt")
+    has_pyproject_toml = await check_path_in_workdir(container, "pyproject.toml")
+
+    if has_pyproject_toml:
+        container = _install_python_dependencies_from_poetry(container)
 
     if has_setup_py:
-        container = container.with_exec(install_connector_package_cmd)
-    if has_requirements_txt:
-        container = container.with_exec(install_requirements_cmd)
+        container = _install_python_dependencies_from_setup_py(container, additional_dependency_groups)
 
-    if additional_dependency_groups:
-        container = container.with_exec(
-            install_connector_package_cmd[:-1] + [install_connector_package_cmd[-1] + f"[{','.join(additional_dependency_groups)}]"]
-        )
+    if has_requirements_txt:
+        container = _install_python_dependencies_from_requirements_txt(container)
 
     return container
 

airbyte-integrations/connectors/destination-duckdb/.dockerignore

@@ -3,3 +3,6 @@
 !main.py
 !destination_duckdb
 !setup.py
+!pyproject.toml
+!poetry.lock
+!README.md

airbyte-integrations/connectors/destination-duckdb/.gitignore

@@ -0,0 +1,2 @@
+# Ignore symlinks created within the dev container
+.symlinks

airbyte-integrations/connectors/destination-duckdb/Dockerfile

@@ -1,42 +1,34 @@
-FROM python:3.9.11 as base
+FROM python:3.9.11
 # FROM python:3.9.11-alpine3.15 as base
 # switched from alpine as there were tons of errors (in case you want to switch back to alpine)
 # - https://stackoverflow.com/a/57485724/5246670
 # - numpy error: https://stackoverflow.com/a/22411624/5246670
 # - libstdc++ https://github.com/amancevice/docker-pandas/issues/12#issuecomment-717215043
 # - musl-dev linux-headers g++ because of: https://stackoverflow.com/a/40407099/5246670
 
-# build and load all requirements
-FROM base as builder
 WORKDIR /airbyte/integration_code
 
-# upgrade pip to the latest version
+# Upgrade system packages and install Poetry
 RUN apt-get update && apt-get -y upgrade \
-    && pip install --upgrade pip
+    && pip install --upgrade pip \
+    && pip install poetry
 
-COPY setup.py ./
-# install necessary packages to a temporary folder
-RUN pip install --prefix=/install .
-# build a clean environment
-FROM base
-# RUN conda install -c conda-forge python-duckdb
-WORKDIR /airbyte/integration_code
-
-# copy all loaded and built libraries to a pure basic image
-COPY --from=builder /install /usr/local
-# add default timezone settings
-COPY --from=builder /usr/share/zoneinfo/Etc/UTC /etc/localtime
-RUN echo "Etc/UTC" > /etc/timezone
+# Copy only poetry.lock* in case it doesn't exist
+COPY pyproject.toml poetry.lock* ./
+RUN poetry config virtualenvs.create false \
+    && poetry install --no-root --no-dev
 
-#adding duckdb manually (outside of setup.py - lots of errors)
-RUN pip install duckdb
-
-# copy payload code only
+# Copy code
 COPY main.py ./
 COPY destination_duckdb ./destination_duckdb
 
+# Timezone setup
+COPY --from=python:3.9.11 /usr/share/zoneinfo/Etc/UTC /etc/localtime
+RUN echo "Etc/UTC" > /etc/timezone
+
+# Entry point
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.0
+LABEL io.airbyte.version=0.2.0
 LABEL io.airbyte.name=airbyte/destination-duckdb

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

@@ -0,0 +1,10 @@
+connector_image: airbyte/destination-duckdb:dev
+acceptance_tests:
+  spec:
+    tests:
+      - spec_path: integration_tests/spec.json
+        config_path: "integration_tests/config.json"
+  connection:
+    tests:
+      - config_path: "integration_tests/config.json"
+        status: "succeed"

airbyte-integrations/connectors/destination-duckdb/destination_duckdb/destination.py

@@ -25,6 +25,9 @@ def _get_destination_path(destination_path: str) -> str:
         Get a normalized version of the destination path.
         Automatically append /local/ to the start of the path
         """
+        if destination_path.startswith("md:") or destination_path.startswith("motherduck:"):
+            return destination_path
+
         if not destination_path.startswith("/local"):
             destination_path = os.path.join("/local", destination_path)
 
@@ -37,9 +40,11 @@ def _get_destination_path(destination_path: str) -> str:
         return destination_path
 
     def write(
-        self, config: Mapping[str, Any], configured_catalog: ConfiguredAirbyteCatalog, input_messages: Iterable[AirbyteMessage]
+        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.
 
@@ -56,17 +61,19 @@ def write(
         streams = {s.stream.name for s in configured_catalog.streams}
         logger.info(f"Starting write to DuckDB with {len(streams)} streams")
 
-        path = config.get("destination_path")
+        path = str(config.get("destination_path"))
         path = self._get_destination_path(path)
-        # check if file exists
 
-        logger.info(f"Opening DuckDB file at {path}")
+        # Get and register auth token if applicable
+        motherduck_api_key = str(config.get("motherduck_api_key"))
+        if motherduck_api_key:
+            os.environ["motherduck_token"] = motherduck_api_key
+
         con = duckdb.connect(database=path, read_only=False)
 
         # create the tables if needed
         # con.execute("BEGIN TRANSACTION")
         for configured_stream in configured_catalog.streams:
-
             name = configured_stream.stream.name
             table_name = f"_airbyte_raw_{name}"
             if configured_stream.destination_sync_mode == DestinationSyncMode.overwrite:
@@ -82,7 +89,7 @@ def write(
             query = f"""
             CREATE TABLE IF NOT EXISTS {table_name} (
                 _airbyte_ab_id TEXT PRIMARY KEY,
-                _airbyte_emitted_at JSON,
+                _airbyte_emitted_at DATETIME,
                 _airbyte_data JSON
             )
             """
@@ -92,20 +99,16 @@ def write(
         buffer = defaultdict(list)
 
         for message in input_messages:
-
             if message.type == Type.STATE:
                 # flush the buffer
                 for stream_name in buffer.keys():
-
                     logger.info(f"flushing buffer for state: {message}")
                     query = """
-                    INSERT INTO {table_name}
+                    INSERT INTO {table_name} (_airbyte_ab_id, _airbyte_emitted_at, _airbyte_data)
                     VALUES (?,?,?)
                     """.format(
                         table_name=f"_airbyte_raw_{stream_name}"
                     )
-                    logger.info(f"query: {query}")
-
                     con.executemany(query, buffer[stream_name])
 
                 con.commit()
@@ -120,13 +123,18 @@ def write(
                     continue
 
                 # add to buffer
-                buffer[stream].append((str(uuid.uuid4()), datetime.datetime.now().isoformat(), json.dumps(data)))
+                buffer[stream].append(
+                    (
+                        str(uuid.uuid4()),
+                        datetime.datetime.now().isoformat(),
+                        json.dumps(data),
+                    )
+                )
             else:
                 logger.info(f"Message type {message.type} not supported, skipping")
 
         # flush any remaining messages
         for stream_name in buffer.keys():
-
             query = """
             INSERT INTO {table_name}
             VALUES (?,?,?)
@@ -150,11 +158,16 @@ def check(self, logger: AirbyteLogger, config: Mapping[str, Any]) -> AirbyteConn
         :return: AirbyteConnectionStatus indicating a Success or Failure
         """
         try:
-            # parse the destination path
-            param_path = config.get("destination_path")
-            path = self._get_destination_path(param_path)
+            path = config.get("destination_path")
+            path = self._get_destination_path(path)
+
+            if path.startswith("/local"):
+                logger.info(f"Using DuckDB file at {path}")
+                os.makedirs(os.path.dirname(path), exist_ok=True)
+
+            if "motherduck_api_key" in config:
+                os.environ["motherduck_token"] = config["motherduck_api_key"]
 
-            os.makedirs(os.path.dirname(path), exist_ok=True)
             con = duckdb.connect(database=path, read_only=False)
             con.execute("SELECT 1;")
 

airbyte-integrations/connectors/destination-duckdb/destination_duckdb/spec.json

@@ -11,10 +11,15 @@
     "required": ["destination_path"],
     "additionalProperties": true,
     "properties": {
+      "motherduck_api_key": {
+        "type": "string",
+        "description": "API key to use for authentication to a MotherDuck database.",
+        "airbyte_secret": true
+      },
       "destination_path": {
         "type": "string",
         "description": "Path to the .duckdb file. The file will be placed inside that local mount. For more information check out our <a href=\"https://docs.airbyte.io/integrations/destinations/duckdb\">docs</a>",
-        "example": "/local/destination.duckdb"
+        "examples": ["/local/destination.duckdb", "md:", "motherduck:"]
       },
       "schema": {
         "type": "string",