Merge branch 'main' into catalog-fixes

Author: aaronsteers

.github/workflows/python_pytest.yml

@@ -48,16 +48,6 @@ jobs:
         --durations=5 --exitfirst
         -m "not slow and not requires_creds and not linting and not flaky"
 
-    - name: Run Pytest with Coverage (Flaky Tests Only)
-      timeout-minutes: 60
-      continue-on-error: true
-      env:
-        GCP_GSM_CREDENTIALS: ${{ secrets.GCP_GSM_CREDENTIALS }}
-      run: >
-        poetry run coverage run -m pytest
-        --durations=5 --exitfirst
-        -m "flaky and not slow and not requires_creds"
-
     - name: Print Coverage Report
       if: always()
       run: poetry run coverage report
@@ -75,6 +65,13 @@ jobs:
         name: fasttest-coverage
         path: htmlcov/
 
+    - name: Upload logs to GitHub Artifacts
+      if: failure()
+      uses: actions/upload-artifact@v4
+      with:
+        name: pytest-fast-test-logs
+        path: /tmp/airbyte/logs/
+
   pytest-no-creds:
     name: Pytest (No Creds)
     runs-on: ubuntu-latest
@@ -122,6 +119,13 @@ jobs:
         name: nocreds-test-coverage
         path: htmlcov/
 
+    - name: Upload logs to GitHub Artifacts
+      if: failure()
+      uses: actions/upload-artifact@v4
+      with:
+        name: pytest-no-creds-test-logs
+        path: /tmp/airbyte/logs/
+
   pytest:
     name: Pytest (All, Python ${{ matrix.python-version }}, ${{ matrix.os }})
     # Don't run on forks. Run on pushes to main, and on PRs that are not from forks.
@@ -133,7 +137,7 @@ jobs:
         python-version: [
           '3.10',
           '3.11',
-          '3.12',
+          # '3.12', # Commented out: Certain tests don't work in Python 3.12, although PyAirbyte itself does work on 3.12
         ]
         os: [
           Ubuntu,
@@ -189,6 +193,13 @@ jobs:
         name: py${{ matrix.python-version }}-${{ matrix.os }}-test-coverage
         path: htmlcov/
 
+    - name: Upload logs to GitHub Artifacts
+      if: failure()
+      uses: actions/upload-artifact@v4
+      with:
+        name: py${{ matrix.python-version }}-${{ matrix.os }}-test-logs
+        path: /tmp/airbyte/logs/
+
   dependency-analysis:
     name: Dependency Analysis with Deptry
     runs-on: ubuntu-latest

.github/workflows/semantic_pr_check.yml

@@ -47,3 +47,13 @@ jobs:
           # scopes: |
           #   core
           #   ui
+
+      - name: Check for "do not merge" in PR title
+        if: ${{ github.event.pull_request.draft == false }}
+        uses: actions/github-script@v6
+        with:
+          script: |
+            const title = context.payload.pull_request.title.toLowerCase();
+            if (title.includes('do not merge') || title.includes('do-not-merge')) {
+              core.setFailed('PR title contains "do not merge" or "do-not-merge". Please remove this before merging.');
+            }

airbyte/_executors/base.py

@@ -185,6 +185,14 @@ def _cli(self) -> list[str]:
         """
         ...
 
+    def map_cli_args(self, args: list[str]) -> list[str]:
+        """Map CLI args if needed.
+
+        By default, this is a no-op. Subclasses may override this method in order to
+        map CLI args into the format expected by the connector.
+        """
+        return args
+
     def execute(
         self,
         args: list[str],
@@ -195,8 +203,9 @@ def execute(
 
         If stdin is provided, it will be passed to the subprocess as STDIN.
         """
+        mapped_args = self.map_cli_args(args)
         with _stream_from_subprocess(
-            [*self._cli, *args],
+            [*self._cli, *mapped_args],
             stdin=stdin,
         ) as stream_lines:
             yield from stream_lines

airbyte/_executors/declarative.py

@@ -18,6 +18,7 @@
 
 
 if TYPE_CHECKING:
+    from argparse import Namespace
     from collections.abc import Iterator
 
     from airbyte._message_iterators import AirbyteMessageIterator
@@ -102,7 +103,9 @@ def execute(
         """Execute the declarative source."""
         _ = stdin  # Not used
         source_entrypoint = AirbyteEntrypoint(self.declarative_source)
-        parsed_args = source_entrypoint.parse_args(args)
+
+        mapped_args: list[str] = self.map_cli_args(args)
+        parsed_args: Namespace = source_entrypoint.parse_args(mapped_args)
         yield from source_entrypoint.run(parsed_args)
 
     def ensure_installation(self, *, auto_fix: bool = True) -> None:

airbyte/_executors/docker.py

@@ -1,22 +1,33 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 from __future__ import annotations
 
+import logging
 import shutil
+from pathlib import Path
 from typing import NoReturn
 
 from airbyte import exceptions as exc
 from airbyte._executors.base import Executor
 
 
+logger = logging.getLogger("airbyte")
+
+
+DEFAULT_AIRBYTE_CONTAINER_TEMP_DIR = "/airbyte/tmp"
+"""Default temp dir in an Airbyte connector's Docker image."""
+
+
 class DockerExecutor(Executor):
     def __init__(
         self,
         name: str | None = None,
         *,
         executable: list[str],
         target_version: str | None = None,
+        volumes: dict[Path, str] | None = None,
     ) -> None:
         self.executable: list[str] = executable
+        self.volumes: dict[Path, str] = volumes or {}
         name = name or executable[0]
         super().__init__(name=name, target_version=target_version)
 
@@ -56,3 +67,40 @@ def uninstall(self) -> NoReturn:
     def _cli(self) -> list[str]:
         """Get the base args of the CLI executable."""
         return self.executable
+
+    def map_cli_args(self, args: list[str]) -> list[str]:
+        """Map local file paths to the container's volume paths."""
+        new_args = []
+        for arg in args:
+            if Path(arg).exists():
+                # This is a file path and we need to map it to the same file within the
+                # relative path of the file within the container's volume.
+                for local_volume, container_path in self.volumes.items():
+                    if Path(arg).is_relative_to(local_volume):
+                        logger.debug(
+                            f"Found file input path `{arg}` "
+                            f"relative to container-mapped volume: {local_volume}"
+                        )
+                        mapped_path = Path(container_path) / Path(arg).relative_to(local_volume)
+                        logger.debug(f"Mapping `{arg}` -> `{mapped_path}`")
+                        new_args.append(str(mapped_path))
+                        break
+                else:
+                    # No break reached; a volume was found for this file path
+                    logger.warning(
+                        f"File path `{arg}` is not relative to any volume path "
+                        f"in the provided volume mappings: {self.volumes}. "
+                        "The file may not be available to the container at runtime."
+                    )
+                    new_args.append(arg)
+
+            else:
+                new_args.append(arg)
+
+        if args != new_args:
+            logger.debug(
+                f"Mapping local-to-container CLI args: {args} -> {new_args} "
+                f"based upon volume definitions: {self.volumes}"
+            )
+
+        return new_args

airbyte/_executors/util.py

@@ -11,7 +11,7 @@
 
 from airbyte import exceptions as exc
 from airbyte._executors.declarative import DeclarativeExecutor
-from airbyte._executors.docker import DockerExecutor
+from airbyte._executors.docker import DEFAULT_AIRBYTE_CONTAINER_TEMP_DIR, DockerExecutor
 from airbyte._executors.local import PathExecutor
 from airbyte._executors.python import VenvExecutor
 from airbyte._util.meta import which
@@ -42,8 +42,8 @@ def _try_get_source_manifest(
 
     Raises:
         - `PyAirbyteInputError`: If `source_name` is `None`.
-        - `HTTPError`: If fetching the URL was unsuccessful.
-        - `YAMLError`: If parsing the YAML failed.
+        - `AirbyteConnectorInstallationError`: If the registry file cannot be downloaded or if the
+          manifest YAML cannot be parsed.
     """
     if source_name is None:
         raise exc.PyAirbyteInputError(
@@ -62,7 +62,16 @@ def _try_get_source_manifest(
         url=manifest_url,
         headers={"User-Agent": f"PyAirbyte/{get_version()}"},
     )
-    response.raise_for_status()  # Raise HTTPError exception if the download failed
+    try:
+        response.raise_for_status()  # Raise HTTPError exception if the download failed
+    except requests.exceptions.HTTPError as ex:
+        raise exc.AirbyteConnectorInstallationError(
+            message="Failed to download the connector manifest.",
+            context={
+                "manifest_url": manifest_url,
+            },
+        ) from ex
+
     try:
         return cast("dict", yaml.safe_load(response.text))
     except yaml.YAMLError as ex:
@@ -115,7 +124,7 @@ def _get_local_executor(
     )
 
 
-def get_connector_executor(  # noqa: PLR0912, PLR0913, PLR0915 # Too many branches/arugments/statements
+def get_connector_executor(  # noqa: PLR0912, PLR0913, PLR0915 # Too many branches/arguments/statements
     name: str,
     *,
     version: str | None = None,
@@ -217,21 +226,24 @@ def get_connector_executor(  # noqa: PLR0912, PLR0913, PLR0915 # Too many branch
         if ":" not in docker_image:
             docker_image = f"{docker_image}:{version or 'latest'}"
 
-        temp_dir = TEMP_DIR_OVERRIDE or Path(tempfile.gettempdir())
+        host_temp_dir = TEMP_DIR_OVERRIDE or Path(tempfile.gettempdir())
+        container_temp_dir = DEFAULT_AIRBYTE_CONTAINER_TEMP_DIR
 
         local_mount_dir = Path().absolute() / name
         local_mount_dir.mkdir(exist_ok=True)
 
+        volumes = {
+            local_mount_dir: "/local",
+            host_temp_dir: container_temp_dir,
+        }
         docker_cmd = [
             "docker",
             "run",
             "--rm",
             "-i",
-            "--volume",
-            f"{local_mount_dir}:/local/",
-            "--volume",
-            f"{temp_dir}:{temp_dir}",
         ]
+        for local_dir, container_dir in volumes.items():
+            docker_cmd.extend(["--volume", f"{local_dir}:{container_dir}"])
 
         if use_host_network is True:
             docker_cmd.extend(["--network", "host"])
@@ -241,6 +253,7 @@ def get_connector_executor(  # noqa: PLR0912, PLR0913, PLR0915 # Too many branch
         return DockerExecutor(
             name=name,
             executable=docker_cmd,
+            volumes=volumes,
         )
 
     if source_manifest:

airbyte/_util/temp_files.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import json
+import stat
 import tempfile
 import time
 import warnings
@@ -36,6 +37,9 @@ def as_temp_files(files_contents: list[dict | str]) -> Generator[list[str], Any,
                 json.dumps(content) if isinstance(content, dict) else content,
             )
             temp_file.flush()
+            # Grant "read" permission to all users
+            Path(temp_file.name).chmod(stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
+
             # Don't close the file yet (breaks Windows)
             # temp_file.close()
             temp_files.append(temp_file)

airbyte/caches/base.py

@@ -4,7 +4,7 @@
 from __future__ import annotations
 
 from pathlib import Path
-from typing import IO, TYPE_CHECKING, Any, ClassVar, final
+from typing import IO, TYPE_CHECKING, Any, ClassVar, Literal, final
 
 import pandas as pd
 import pyarrow as pa
@@ -34,6 +34,7 @@
     from airbyte.shared.sql_processor import SqlProcessorBase
     from airbyte.shared.state_providers import StateProviderBase
     from airbyte.shared.state_writers import StateWriterBase
+    from airbyte.sources.base import Source
     from airbyte.strategies import WriteStrategy
 
 
@@ -293,6 +294,35 @@ def register_source(
             incoming_stream_names=stream_names,
         )
 
+    def create_source_tables(
+        self,
+        source: Source,
+        streams: Literal["*"] | list[str] | None = None,
+    ) -> None:
+        """Create tables in the cache for the provided source if they do not exist already.
+
+        Tables are created based upon the Source's catalog.
+
+        Args:
+            source: The source to create tables for.
+            streams: Stream names to create tables for. If None, use the Source's selected_streams
+                or "*" if neither is set. If "*", all available streams will be used.
+        """
+        if streams is None:
+            streams = source.get_selected_streams() or "*"
+
+        catalog_provider = CatalogProvider(source.get_configured_catalog(streams=streams))
+
+        # Ensure schema exists
+        self.processor._ensure_schema_exists()  # noqa: SLF001  # Accessing non-public member
+
+        # Create tables for each stream if they don't exist
+        for stream_name in catalog_provider.stream_names:
+            self.processor._ensure_final_table_exists(  # noqa: SLF001
+                stream_name=stream_name,
+                create_if_missing=True,
+            )
+
     def __getitem__(self, stream: str) -> CachedDataset:
         """Return a dataset by stream name."""
         return self.streams[stream]

airbyte/constants.py

@@ -109,3 +109,18 @@ def _str_to_bool(value: str) -> bool:
 This setting helps you make informed choices about data privacy and operation in restricted and
 air-gapped environments.
 """
+
+AIRBYTE_PRINT_FULL_ERROR_LOGS: bool = _str_to_bool(
+    os.getenv(
+        key="AIRBYTE_PRINT_FULL_ERROR_LOGS",
+        default=os.getenv("CI", "false"),
+    )
+)
+"""Whether to print full error logs when an error occurs.
+This setting helps in debugging by providing detailed logs when errors occur. This is especially
+helpful in ephemeral environments like CI/CD pipelines where log files may not be persisted after
+the pipeline run.
+
+If not set, the default value is `False` for non-CI environments.
+If running in a CI environment ("CI" env var is set), then the default value is `True`.
+"""