merge from main

Author: aldogonzalez8

.github/actions/check-docker-tag/action.yml

@@ -0,0 +1,28 @@
+name: 'Check Docker Tag Exists'
+description: 'Check if a Docker tag exists on DockerHub to prevent overwrites'
+inputs:
+  image_name:
+    description: 'Docker image name (e.g. airbyte/source-declarative-manifest)'
+    required: true
+  tag:
+    description: 'Docker tag to check'
+    required: true
+runs:
+  using: "composite"
+  steps:
+    - name: "Check for existing tag (${{ inputs.image_name }}:${{ inputs.tag }})"
+      shell: bash
+      run: |
+        image="${{ inputs.image_name }}"
+        tag_input="${{ inputs.tag }}"
+        if [ -z "$image" ] || [ -z "$tag_input" ]; then
+          echo "Error: image_name and tag are required."
+          exit 1
+        fi
+        tag="${image}:${tag_input}"
+        echo "Checking if tag '$tag' exists on DockerHub..."
+        if DOCKER_CLI_EXPERIMENTAL=enabled docker manifest inspect "$tag" > /dev/null 2>&1; then
+          echo "The tag '$tag' already exists on DockerHub. Skipping publish to prevent overwrite."
+          exit 1
+        fi
+        echo "No existing tag '$tag' found. Proceeding with publish."

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

@@ -1,13 +1,15 @@
 name: Docker Build Check
+permissions:
+  contents: read
 
 on:
   pull_request:
     branches:
       - main
 
 jobs:
-  docker-build-check:
-    name: SDM Docker Image Build # Renamed job to be more descriptive
+  sdm-docker-build-check:
+    name: SDM Docker Image Build
     runs-on: ubuntu-24.04
     steps:
       - name: Checkout code
@@ -42,3 +44,29 @@ jobs:
           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 }}
+
+  manifest-server-docker-build-check:
+    name: Manifest Server Docker Image Build
+    runs-on: ubuntu-24.04
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - 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 Manifest Server Docker image for multiple platforms
+        id: manifest-server-build
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          file: airbyte_cdk/manifest_server/Dockerfile
+          platforms: linux/amd64,linux/arm64
+          push: false
+          tags: airbyte/manifest-server:pr-${{ github.event.pull_request.number }}
+          outputs: type=image,name=target,annotation-index.org.opencontainers.image.description=Manifest Server Docker image for PR ${{ github.event.pull_request.number }}

.github/workflows/pypi_publish.yml renamed to .github/workflows/publish.yml

@@ -6,6 +6,8 @@
 # we have to also update the Trusted Publisher settings on PyPI.
 
 name: CDK Publish
+permissions:
+  contents: read
 
 on:
   push:
@@ -31,6 +33,11 @@ on:
         type: boolean
         required: true
         default: true
+      publish_manifest_server:
+        description: "Publish Manifest Server to DockerHub. If true, the workflow will publish the Manifest Server to DockerHub."
+        type: boolean
+        required: true
+        default: true
       update_connector_builder:
         description: "Update Connector Builder. If true, the workflow will create a PR to bump the CDK version used by Connector Builder."
         type: boolean
@@ -204,18 +211,10 @@ jobs:
 
       - name: "Check for existing tag (version: ${{ env.VERSION || 'none' }} )"
         if: env.VERSION != ''
-        run: |
-          tag="airbyte/source-declarative-manifest:${{ env.VERSION }}"
-          if [ -z "$tag" ]; then
-            echo "Error: VERSION is not set. Ensure the tag follows the format 'refs/tags/vX.Y.Z'."
-            exit 1
-          fi
-          echo "Checking if tag '$tag' exists on DockerHub..."
-          if DOCKER_CLI_EXPERIMENTAL=enabled docker manifest inspect "$tag" > /dev/null 2>&1; then
-            echo "The tag '$tag' already exists on DockerHub. Skipping publish to prevent overwrite."
-            exit 1
-          fi
-          echo "No existing tag '$tag' found. Proceeding with publish."
+        uses: ./.github/actions/check-docker-tag
+        with:
+          image_name: airbyte/source-declarative-manifest
+          tag: ${{ env.VERSION }}
 
       - name: "Build and push (sha tag: '${{ github.sha }}')"
         # Only run if the version is not set
@@ -250,6 +249,90 @@ jobs:
           tags: |
             airbyte/source-declarative-manifest:latest
 
+  publish_manifest_server:
+    name: Publish Manifest Server to DockerHub
+    if: >
+      (github.event_name == 'push' &&
+       startsWith(github.ref, 'refs/tags/v')) ||
+      (github.event_name == 'workflow_dispatch' &&
+       github.event.inputs.publish_manifest_server == 'true'
+      )
+    runs-on: ubuntu-24.04
+    needs: [build]
+    environment:
+      name: DockerHub
+      url: https://hub.docker.com/r/airbyte/manifest-server/tags
+    env:
+      VERSION: ${{ needs.build.outputs.VERSION }}
+      IS_PRERELEASE: ${{ needs.build.outputs.IS_PRERELEASE }}
+
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      # We need to download the build artifact again because the previous job was on a different runner
+      - name: Download Build Artifact
+        uses: actions/download-artifact@v4
+        with:
+          name: Packages-${{ github.run_id }}
+          path: 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: Login to Docker Hub
+        uses: docker/login-action@v3
+        with:
+          username: ${{ secrets.DOCKER_HUB_USERNAME }}
+          password: ${{ secrets.DOCKER_HUB_PASSWORD }}
+
+      - name: "Check for existing tag (version: ${{ env.VERSION || 'none' }} )"
+        if: env.VERSION != ''
+        uses: ./.github/actions/check-docker-tag
+        with:
+          image_name: airbyte/manifest-server
+          tag: ${{ env.VERSION }}
+
+      - name: "Build and push (sha tag: '${{ github.sha }}')"
+        # Only run if the version is not set
+        if: env.VERSION == ''
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          file: airbyte_cdk/manifest_server/Dockerfile
+          platforms: linux/amd64,linux/arm64
+          push: true
+          tags: |
+            airbyte/manifest-server:${{ github.sha }}
+
+      - name: "Build and push (version tag: ${{ env.VERSION || 'none'}})"
+        # Only run if the version is set
+        if: env.VERSION != ''
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          file: airbyte_cdk/manifest_server/Dockerfile
+          platforms: linux/amd64,linux/arm64
+          push: true
+          tags: |
+            airbyte/manifest-server:${{ env.VERSION }}
+
+      - name: Build and push ('latest' tag)
+        # Only run if version is set and IS_PRERELEASE is false
+        if: env.VERSION != '' && env.IS_PRERELEASE == 'false'
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          file: airbyte_cdk/manifest_server/Dockerfile
+          platforms: linux/amd64,linux/arm64
+          push: true
+          tags: |
+            airbyte/manifest-server:latest
+
   update-connector-builder:
     # Create a PR against the Builder, to update the CDK version that it uses.
     # In the future, Builder may use the SDM docker image instead of the Python CDK package.

airbyte_cdk/__init__.py

@@ -107,7 +107,6 @@
 from .sources.declarative.extractors.record_filter import RecordFilter
 from .sources.declarative.incremental import DatetimeBasedCursor
 from .sources.declarative.interpolation import InterpolatedBoolean, InterpolatedString
-from .sources.declarative.manifest_declarative_source import ManifestDeclarativeSource
 from .sources.declarative.migrations.legacy_to_per_partition_state_migration import (
     LegacyToPerPartitionStateMigration,
 )
@@ -253,7 +252,6 @@
     "JsonDecoder",
     "JsonFileSchemaLoader",
     "LegacyToPerPartitionStateMigration",
-    "ManifestDeclarativeSource",
     "MinMaxDatetime",
     "NoAuth",
     "OffsetIncrement",

airbyte_cdk/connector_builder/connector_builder_handler.py

@@ -19,8 +19,6 @@
     ConcurrentDeclarativeSource,
     TestLimits,
 )
-from airbyte_cdk.sources.declarative.declarative_source import DeclarativeSource
-from airbyte_cdk.sources.declarative.manifest_declarative_source import ManifestDeclarativeSource
 from airbyte_cdk.utils.airbyte_secrets_utils import filter_secrets
 from airbyte_cdk.utils.datetime_helpers import ab_datetime_now
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
@@ -90,7 +88,7 @@ def create_source(
 
 
 def read_stream(
-    source: DeclarativeSource,
+    source: ConcurrentDeclarativeSource[Optional[List[AirbyteStateMessage]]],
     config: Mapping[str, Any],
     configured_catalog: ConfiguredAirbyteCatalog,
     state: List[AirbyteStateMessage],
@@ -128,7 +126,9 @@ def read_stream(
         return error.as_airbyte_message()
 
 
-def resolve_manifest(source: ManifestDeclarativeSource) -> AirbyteMessage:
+def resolve_manifest(
+    source: ConcurrentDeclarativeSource[Optional[List[AirbyteStateMessage]]],
+) -> AirbyteMessage:
     try:
         return AirbyteMessage(
             type=Type.RECORD,
@@ -145,7 +145,9 @@ def resolve_manifest(source: ManifestDeclarativeSource) -> AirbyteMessage:
         return error.as_airbyte_message()
 
 
-def full_resolve_manifest(source: ManifestDeclarativeSource, limits: TestLimits) -> AirbyteMessage:
+def full_resolve_manifest(
+    source: ConcurrentDeclarativeSource[Optional[List[AirbyteStateMessage]]], limits: TestLimits
+) -> AirbyteMessage:
     try:
         manifest = {**source.resolved_manifest}
         streams = manifest.get("streams", [])

airbyte_cdk/connector_builder/main.py

@@ -25,7 +25,9 @@
     ConfiguredAirbyteCatalog,
     ConfiguredAirbyteCatalogSerializer,
 )
-from airbyte_cdk.sources.declarative.manifest_declarative_source import ManifestDeclarativeSource
+from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
+    ConcurrentDeclarativeSource,
+)
 from airbyte_cdk.sources.source import Source
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 
@@ -68,7 +70,7 @@ def get_config_and_catalog_from_args(
 
 
 def handle_connector_builder_request(
-    source: ManifestDeclarativeSource,
+    source: ConcurrentDeclarativeSource[Optional[List[AirbyteStateMessage]]],
     command: str,
     config: Mapping[str, Any],
     catalog: Optional[ConfiguredAirbyteCatalog],

airbyte_cdk/connector_builder/test_reader/reader.py

@@ -23,7 +23,9 @@
     ConfiguredAirbyteCatalog,
     TraceType,
 )
-from airbyte_cdk.sources.declarative.declarative_source import DeclarativeSource
+from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
+    ConcurrentDeclarativeSource,
+)
 from airbyte_cdk.utils import AirbyteTracedException
 from airbyte_cdk.utils.datetime_format_inferrer import DatetimeFormatInferrer
 from airbyte_cdk.utils.schema_inferrer import (
@@ -55,7 +57,7 @@ class TestReader:
             that contains slices of data, log messages, auxiliary requests, and any inferred schema or datetime formats.
 
             Parameters:
-                source (DeclarativeSource): The data source to read from.
+                source (ConcurrentDeclarativeSource): The data source to read from.
                 config (Mapping[str, Any]): Configuration parameters for the source.
                 configured_catalog (ConfiguredAirbyteCatalog): Catalog containing stream configuration.
                 state (List[AirbyteStateMessage]): Current state information for the read.
@@ -83,7 +85,7 @@ def __init__(
 
     def run_test_read(
         self,
-        source: DeclarativeSource,
+        source: ConcurrentDeclarativeSource[Optional[List[AirbyteStateMessage]]],
         config: Mapping[str, Any],
         configured_catalog: ConfiguredAirbyteCatalog,
         stream_name: str,
@@ -94,7 +96,7 @@ def run_test_read(
         Run a test read for the connector by reading from a single stream and inferring schema and datetime formats.
 
         Parameters:
-            source (DeclarativeSource): The source instance providing the streams.
+            source (ConcurrentDeclarativeSource): The source instance providing the streams.
             config (Mapping[str, Any]): The configuration settings to use for reading.
             configured_catalog (ConfiguredAirbyteCatalog): The catalog specifying the stream configuration.
             state (List[AirbyteStateMessage]): A list of state messages to resume the read.
@@ -126,7 +128,7 @@ def run_test_read(
             if stream
             else None,
             self._cursor_field_to_nested_and_composite_field(stream.cursor_field)
-            if stream
+            if stream and stream.cursor_field
             else None,
         )
         datetime_format_inferrer = DatetimeFormatInferrer()
@@ -381,13 +383,13 @@ def _get_latest_config_update(
 
     def _read_stream(
         self,
-        source: DeclarativeSource,
+        source: ConcurrentDeclarativeSource[Optional[List[AirbyteStateMessage]]],
         config: Mapping[str, Any],
         configured_catalog: ConfiguredAirbyteCatalog,
         state: List[AirbyteStateMessage],
     ) -> Iterator[AirbyteMessage]:
         """
-        Reads messages from the given DeclarativeSource using an AirbyteEntrypoint.
+        Reads messages from the given ConcurrentDeclarativeSource using an AirbyteEntrypoint.
 
         This method attempts to yield messages from the source's read generator. If the generator
         raises an AirbyteTracedException, it checks whether the exception message indicates a non-actionable
@@ -396,7 +398,7 @@ def _read_stream(
         wrapped into an AirbyteTracedException, and yielded as an AirbyteMessage.
 
         Parameters:
-            source (DeclarativeSource): The source object that provides data reading logic.
+            source (ConcurrentDeclarativeSource): The source object that provides data reading logic.
             config (Mapping[str, Any]): The configuration dictionary for the source.
             configured_catalog (ConfiguredAirbyteCatalog): The catalog defining the streams and their configurations.
             state (List[AirbyteStateMessage]): A list representing the current state for incremental sync.

airbyte_cdk/legacy/__init__.py

@@ -0,0 +1 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.

airbyte_cdk/legacy/sources/__init__.py

@@ -0,0 +1 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.

airbyte_cdk/legacy/sources/declarative/__init__.py

@@ -0,0 +1 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.