Merge remote-tracking branch 'origin/main' into aj/feat/mini-cat-test-suites

Author: aaronsteers

.github/workflows/docker-build-check.yml

@@ -0,0 +1,44 @@
+name: Docker Build Check
+
+on:
+  pull_request:
+    branches:
+      - main
+
+jobs:
+  docker-build-check:
+    name: SDM Docker Image Build  # Renamed job to be more descriptive
+    runs-on: ubuntu-24.04
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      # Build the Python package to create the wheel files needed for the Docker build
+      - name: Build Python Package
+        uses: hynek/build-and-inspect-python-package@v2
+        env:
+          POETRY_DYNAMIC_VERSIONING_BYPASS: "0.0.0dev0"
+
+      # Copy the wheel files to the dist directory
+      - name: Copy wheel files to dist directory
+        run: |
+          mkdir -p dist
+          cp /tmp/baipp/dist/*.whl dist/
+
+      - name: Set up QEMU for multi-platform builds
+        uses: docker/setup-qemu-action@v3
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Build Docker image for multiple platforms
+        id: docker-build
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          platforms: linux/amd64,linux/arm64
+          push: false
+          tags: airbyte/source-declarative-manifest:pr-${{ github.event.pull_request.number }}
+          outputs: type=image,name=target,annotation-index.org.opencontainers.image.description=SDM Docker image for PR ${{ github.event.pull_request.number }}

.github/workflows/python_dependency_analysis.yml

@@ -0,0 +1,38 @@
+name: Dependency Analysis
+
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - "airbyte_cdk/**"
+      - "poetry.lock"
+      - "pyproject.toml"
+  pull_request:
+    paths:
+      - "airbyte_cdk/**"
+      - "poetry.lock"
+      - "pyproject.toml"
+
+jobs:
+  dependency-analysis:
+    name: Dependency Analysis with Deptry
+    runs-on: ubuntu-24.04
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+    - name: Set up Python
+      uses: actions/setup-python@v5
+      with:
+        python-version: '3.10'
+    - name: Set up Poetry
+      uses: Gr1N/setup-poetry@v9
+      with:
+        poetry-version: "2.0.1"
+    - name: Install dependencies
+      run: poetry install --all-extras
+
+    # Job-specific step(s):
+    - name: Run Deptry
+      run: |
+        poetry run deptry .

.github/workflows/semantic_pr_check.yml

@@ -39,3 +39,13 @@ jobs:
             Build
             tests
             Tests
+            
+      - 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.');
+            }

Dockerfile

@@ -5,7 +5,7 @@
 # A new version of source-declarative-manifest is built for every new Airbyte CDK release, and their versions are kept in sync.
 #
 
-FROM docker.io/airbyte/python-connector-base:3.0.0@sha256:1a0845ff2b30eafa793c6eee4e8f4283c2e52e1bbd44eed6cb9e9abd5d34d844
+FROM docker.io/airbyte/python-connector-base:4.0.0@sha256:d9894b6895923b379f3006fa251147806919c62b7d9021b5cd125bb67d7bbe22
 
 WORKDIR /airbyte/integration_code
 
@@ -24,8 +24,8 @@ RUN pip install dist/*.whl
 RUN mkdir -p source_declarative_manifest \
     && echo 'from source_declarative_manifest.run import run\n\nif __name__ == "__main__":\n    run()' > main.py \
     && touch source_declarative_manifest/__init__.py \
-    && cp /usr/local/lib/python3.10/site-packages/airbyte_cdk/cli/source_declarative_manifest/_run.py source_declarative_manifest/run.py \
-    && cp /usr/local/lib/python3.10/site-packages/airbyte_cdk/cli/source_declarative_manifest/spec.json source_declarative_manifest/
+    && cp /usr/local/lib/python3.11/site-packages/airbyte_cdk/cli/source_declarative_manifest/_run.py source_declarative_manifest/run.py \
+    && cp /usr/local/lib/python3.11/site-packages/airbyte_cdk/cli/source_declarative_manifest/spec.json source_declarative_manifest/
 
 # Remove unnecessary build files
 RUN rm -rf dist/ pyproject.toml poetry.lock README.md

airbyte_cdk/connector_builder/connector_builder_handler.py

@@ -4,7 +4,7 @@
 
 
 from dataclasses import asdict, dataclass, field
-from typing import Any, List, Mapping
+from typing import Any, Dict, List, Mapping
 
 from airbyte_cdk.connector_builder.test_reader import TestReader
 from airbyte_cdk.models import (
@@ -27,30 +27,34 @@
 DEFAULT_MAXIMUM_NUMBER_OF_PAGES_PER_SLICE = 5
 DEFAULT_MAXIMUM_NUMBER_OF_SLICES = 5
 DEFAULT_MAXIMUM_RECORDS = 100
+DEFAULT_MAXIMUM_STREAMS = 100
 
 MAX_PAGES_PER_SLICE_KEY = "max_pages_per_slice"
 MAX_SLICES_KEY = "max_slices"
 MAX_RECORDS_KEY = "max_records"
+MAX_STREAMS_KEY = "max_streams"
 
 
 @dataclass
-class TestReadLimits:
+class TestLimits:
     max_records: int = field(default=DEFAULT_MAXIMUM_RECORDS)
     max_pages_per_slice: int = field(default=DEFAULT_MAXIMUM_NUMBER_OF_PAGES_PER_SLICE)
     max_slices: int = field(default=DEFAULT_MAXIMUM_NUMBER_OF_SLICES)
+    max_streams: int = field(default=DEFAULT_MAXIMUM_STREAMS)
 
 
-def get_limits(config: Mapping[str, Any]) -> TestReadLimits:
+def get_limits(config: Mapping[str, Any]) -> TestLimits:
     command_config = config.get("__test_read_config", {})
     max_pages_per_slice = (
         command_config.get(MAX_PAGES_PER_SLICE_KEY) or DEFAULT_MAXIMUM_NUMBER_OF_PAGES_PER_SLICE
     )
     max_slices = command_config.get(MAX_SLICES_KEY) or DEFAULT_MAXIMUM_NUMBER_OF_SLICES
     max_records = command_config.get(MAX_RECORDS_KEY) or DEFAULT_MAXIMUM_RECORDS
-    return TestReadLimits(max_records, max_pages_per_slice, max_slices)
+    max_streams = command_config.get(MAX_STREAMS_KEY) or DEFAULT_MAXIMUM_STREAMS
+    return TestLimits(max_records, max_pages_per_slice, max_slices, max_streams)
 
 
-def create_source(config: Mapping[str, Any], limits: TestReadLimits) -> ManifestDeclarativeSource:
+def create_source(config: Mapping[str, Any], limits: TestLimits) -> ManifestDeclarativeSource:
     manifest = config["__injected_declarative_manifest"]
     return ManifestDeclarativeSource(
         config=config,
@@ -71,7 +75,7 @@ def read_stream(
     config: Mapping[str, Any],
     configured_catalog: ConfiguredAirbyteCatalog,
     state: List[AirbyteStateMessage],
-    limits: TestReadLimits,
+    limits: TestLimits,
 ) -> AirbyteMessage:
     try:
         test_read_handler = TestReader(
@@ -117,5 +121,40 @@ def resolve_manifest(source: ManifestDeclarativeSource) -> AirbyteMessage:
         return error.as_airbyte_message()
 
 
+def full_resolve_manifest(source: ManifestDeclarativeSource, limits: TestLimits) -> AirbyteMessage:
+    try:
+        manifest = {**source.resolved_manifest}
+        streams = manifest.get("streams", [])
+        for stream in streams:
+            stream["dynamic_stream_name"] = None
+
+        mapped_streams: Dict[str, List[Dict[str, Any]]] = {}
+        for stream in source.dynamic_streams:
+            generated_streams = mapped_streams.setdefault(stream["dynamic_stream_name"], [])
+
+            if len(generated_streams) < limits.max_streams:
+                generated_streams += [stream]
+
+        for generated_streams_list in mapped_streams.values():
+            streams.extend(generated_streams_list)
+
+        manifest["streams"] = streams
+        return AirbyteMessage(
+            type=Type.RECORD,
+            record=AirbyteRecordMessage(
+                data={"manifest": manifest},
+                emitted_at=_emitted_at(),
+                stream="full_resolve_manifest",
+            ),
+        )
+    except AirbyteTracedException as exc:
+        return exc.as_airbyte_message()
+    except Exception as exc:
+        error = AirbyteTracedException.from_exception(
+            exc, message=f"Error full resolving manifest: {str(exc)}"
+        )
+        return error.as_airbyte_message()
+
+
 def _emitted_at() -> int:
     return ab_datetime_now().to_epoch_millis()

airbyte_cdk/connector_builder/main.py

@@ -10,8 +10,9 @@
 
 from airbyte_cdk.connector import BaseConnector
 from airbyte_cdk.connector_builder.connector_builder_handler import (
-    TestReadLimits,
+    TestLimits,
     create_source,
+    full_resolve_manifest,
     get_limits,
     read_stream,
     resolve_manifest,
@@ -72,7 +73,7 @@ def handle_connector_builder_request(
     config: Mapping[str, Any],
     catalog: Optional[ConfiguredAirbyteCatalog],
     state: List[AirbyteStateMessage],
-    limits: TestReadLimits,
+    limits: TestLimits,
 ) -> AirbyteMessage:
     if command == "resolve_manifest":
         return resolve_manifest(source)
@@ -81,6 +82,8 @@ def handle_connector_builder_request(
             catalog is not None
         ), "`test_read` requires a valid `ConfiguredAirbyteCatalog`, got None."
         return read_stream(source, config, catalog, state, limits)
+    elif command == "full_resolve_manifest":
+        return full_resolve_manifest(source, limits)
     else:
         raise ValueError(f"Unrecognized command {command}.")
 

airbyte_cdk/connector_builder/models.py

@@ -21,20 +21,6 @@ class HttpRequest:
     body: Optional[str] = None
 
 
-@dataclass
-class StreamReadPages:
-    records: List[object]
-    request: Optional[HttpRequest] = None
-    response: Optional[HttpResponse] = None
-
-
-@dataclass
-class StreamReadSlices:
-    pages: List[StreamReadPages]
-    slice_descriptor: Optional[Dict[str, Any]]
-    state: Optional[List[Dict[str, Any]]] = None
-
-
 @dataclass
 class LogMessage:
     message: str
@@ -46,11 +32,27 @@ class LogMessage:
 @dataclass
 class AuxiliaryRequest:
     title: str
+    type: str
     description: str
     request: HttpRequest
     response: HttpResponse
 
 
+@dataclass
+class StreamReadPages:
+    records: List[object]
+    request: Optional[HttpRequest] = None
+    response: Optional[HttpResponse] = None
+
+
+@dataclass
+class StreamReadSlices:
+    pages: List[StreamReadPages]
+    slice_descriptor: Optional[Dict[str, Any]]
+    state: Optional[List[Dict[str, Any]]] = None
+    auxiliary_requests: Optional[List[AuxiliaryRequest]] = None
+
+
 @dataclass
 class StreamRead(object):
     logs: List[LogMessage]