airbyte-source-declarative-manifest

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()