Airbyte Declarative Manifest Source

6.5.2 · active · verified Thu Apr 16

This library serves as the base source implementation for low-code Airbyte connectors, allowing users to define data sources using a YAML-based declarative manifest. It interprets these manifests to run data synchronization processes without requiring extensive Python coding. The current version is 6.5.2, and it follows the Airbyte platform's rapid release cadence, with major platform updates typically monthly, influencing the underlying CDK and manifest capabilities.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to programmatically use the `DeclarativeSource` from `airbyte-cdk` to load and interact with a declarative manifest. This is how the `airbyte-source-declarative-manifest` library is implicitly used. It defines a minimal YAML manifest, saves it, and then runs a 'check' operation against it. In a real Airbyte deployment, the platform would handle loading the manifest and invoking the source.

import json
import os
from pathlib import Path

from airbyte_cdk.sources.declarative.declarative_source import DeclarativeSource

# Create a dummy manifest file for demonstration
manifest_content = """
version: "0.3.0"
resolver:
  type: "inline"
schemas:
  streams:
    - name: "example_stream"
      retriever:
        type: "simple"
        requester:
          url_base: "https://jsonplaceholder.typicode.com"
          path: "/todos/1"
          http_method: "GET"
          authenticator:
            type: "no_auth"
        record_selector:
          field_path: []
      stream_mapper:
        type: "simple"
      primary_key: []
      transformations: []
"""

# Save the manifest to a temporary file
manifest_path = Path("temp_manifest.yaml")
manifest_path.write_text(manifest_content)

# Define the config for the source (can be empty if not needed by manifest)
config = {}

# Initialize the declarative source
source = DeclarativeSource(str(manifest_path))

# Run a check operation (or other CDK operations like discover, read)
# This is how Airbyte internally invokes the source.
status = source.check(None, config)

print(f"Source check status: {status.status.value}")

# Clean up the temporary manifest file
manifest_path.unlink()

view raw JSON →