conduit-core 1.0.0

The dbt of data ingestion - declarative, reliable, and testable data pipelines

Conduit Core

Declarative data ingestion framework for Python — inspired by dbt's approach to analytics engineering.

Conduit Core is an open-source Python framework for building production-grade data pipelines. Move data between CSV, JSON, Parquet, PostgreSQL, MySQL, S3, Snowflake, and BigQuery with YAML configuration, automatic retries, checkpoints, and comprehensive testing.

Features

• Declarative YAML Configuration — Define sources, destinations, and transformations without writing Python
• 8 Connectors — CSV, JSON, Parquet, S3, PostgreSQL, MySQL, Snowflake, BigQuery
• Production-Ready Reliability — Atomic operations, automatic retries, dead letter queue, checkpoint/resume
• Schema Management — Auto-inference, validation, evolution tracking, and quality checks
• Developer Experience — Rich progress bars, dry-run mode, audit trail, and comprehensive testing (141+ tests)

Quick Start

pip install conduit-core

Create ingest.yml:

sources:
  - name: users_csv
    type: csv
    path: "./users.csv"

destinations:
  - name: users_json
    type: json
    path: "./users.json"

resources:
  - name: csv_to_json
    source: users_csv
    destination: users_json

Run:

conduit run csv_to_json

Documentation

• Installation & Setup — Installation, environment setup, and credentials
• Usage Guide — CLI commands, pipeline configuration, and examples
• Connectors — Configuration for all 8 connectors
• Features — Reliability, checkpoints, DLQ, incremental loads
• Architecture — Design patterns, tech stack, and structure
• Data Quality — Schema validation and quality checks
• Contributing — Development setup and contribution guidelines
• Roadmap — Future features and connector expansion

Example: Incremental PostgreSQL → Snowflake

sources:
  - name: pg_source
    type: postgresql
    host: ${PG_HOST}
    database: analytics
    user: ${PG_USER}
    password: ${PG_PASSWORD}

destinations:
  - name: sf_dest
    type: snowflake
    account: ${SF_ACCOUNT}
    warehouse: COMPUTE_WH
    database: ANALYTICS
    schema: PUBLIC

resources:
  - name: incremental_load
    source: pg_source
    destination: sf_dest
    query: "SELECT * FROM events"
    incremental_column: created_at
    write_mode: append

Run with validation:

conduit preflight incremental_load
conduit run incremental_load

Why Conduit Core?

Built for data engineers who need the simplicity of declarative configuration with the reliability of enterprise-grade systems:

• ✅ No boilerplate — YAML instead of Python for 90% of use cases
• ✅ Crash-safe — Atomic writes, transactions, and checkpoint/resume
• ✅ Observable — Audit trail, progress bars, and detailed error messages
• ✅ Extensible — Plugin architecture for custom connectors
• ✅ Battle-tested — 141+ tests covering edge cases and failure modes

Community

• Issues & Bugs: GitHub Issues
• Feature Requests: GitHub Discussions
• Contributing: See CONTRIBUTING.md

License

BSL 1.1 (Business Source License) — See LICENSE for details