OpenIngest

Configuration-driven data ingestion framework for Python.

OpenIngest is an open-source, production-grade data ingestion framework that replaces hand-written ETL scripts with a single YAML configuration file. Register a dataset once and OpenIngest handles discovery → schema validation → data quality → transformation → PostgreSQL loading → metadata logging → Airflow DAG generation → notifications — automatically.

v3.0 ships 17 connectors, a full YAML transformation engine, 93 tests, and is published on PyPI.

Install

pip install openingest

Install extras for specific connector types:

pip install openingest[s3]            # Amazon S3
pip install openingest[azure]         # Azure Blob Storage
pip install openingest[gcs]           # Google Cloud Storage
pip install openingest[mysql]         # MySQL
pip install openingest[mongodb]       # MongoDB
pip install openingest[sftp]          # SFTP
pip install openingest[salesforce]    # Salesforce
pip install openingest[hubspot]       # HubSpot
pip install openingest[stripe]        # Stripe
pip install openingest[google_sheets] # Google Sheets
pip install openingest[connectors]    # All v3.0 connectors
pip install openingest[all]           # Everything

Quick Start

# 1. Install
pip install openingest

# 2. Scaffold a project
openingest init my-pipeline
cd my-pipeline

# 3. Set your database URL in .env
# DATABASE_URL=postgresql://user:password@localhost:5432/openingest

# 4. Start PostgreSQL
docker compose up -d

# 5. Infer config from a CSV and run
openingest infer data/raw/orders.csv
openingest run

Output:

================================================================================
OPENINGEST
================================================================================
Run ID : OI-20260703-3BB09C
================================================================================

================================================================================
ORDERS
================================================================================
Quality Check : PASS (98.50%)

================================================================================
PIPELINE SUMMARY
================================================================================
Run ID            : OI-20260703-3BB09C
Datasets Found    : 3
Processed         : 3
Rows Loaded       : 174,777
Duration          : 4.21 sec
Status            : SUCCESS
================================================================================

Dataset Configuration

Everything lives in configs/datasets.yaml. No Python required.

Local CSV

customers:
  file: customers.csv
  staging_table: stg_customers
  load_strategy: replace
  primary_key: [customer_id]
  required_columns: [customer_id, name, email]
  non_null_columns: [customer_id, email]
  unique_columns: [customer_id]

Incremental load with CDC

orders:
  file: orders.csv
  staging_table: stg_orders
  load_strategy: incremental
  incremental_column: order_time
  primary_key: [order_id]
  hash_columns: [customer_id, total_usd, payment_method]
  required_columns: [order_id, customer_id, order_time, total_usd]

Amazon S3

orders_s3:
  source:
    type: s3
    bucket: my-bucket
    key: orders/orders.parquet
    region: us-east-1
    aws_access_key_id: ${AWS_ACCESS_KEY_ID}
    aws_secret_access_key: ${AWS_SECRET_ACCESS_KEY}
  staging_table: stg_orders
  load_strategy: incremental
  incremental_column: order_time
  primary_key: [order_id]

PostgreSQL source (v3.0)

pg_orders:
  source:
    type: postgresql
    host: ${PG_HOST}
    database: ${PG_DATABASE}
    username: ${PG_USER}
    password: ${PG_PASSWORD}
    query: "SELECT * FROM orders WHERE status = 'active'"
  staging_table: stg_pg_orders
  load_strategy: incremental
  incremental_column: updated_at
  primary_key: [order_id]

Salesforce (v3.0)

sf_opportunities:
  source:
    type: salesforce
    username: ${SF_USERNAME}
    password: ${SF_PASSWORD}
    security_token: ${SF_SECURITY_TOKEN}
    client_id: ${SF_CLIENT_ID}
    client_secret: ${SF_CLIENT_SECRET}
    object: Opportunity
    fields: [Id, Name, Amount, StageName, CloseDate]
  staging_table: stg_sf_opportunities
  load_strategy: replace

Transformations (v3.0)

air_data:
  source:
    type: s3
    bucket: my-bucket
    key: Air_full-Raw.csv
    aws_access_key_id: ${AWS_ACCESS_KEY_ID}
    aws_secret_access_key: ${AWS_SECRET_ACCESS_KEY}
    region: ap-south-1
  staging_table: stg_air_clean
  load_strategy: replace
  transforms:
    - type: rename
      columns:
        "PM2.5": pm25
        "NO2(GT)": no2
    - type: cast
      columns:
        pm25: float
        no2: float
    - type: filter
      expression: "pm25 >= 0"
    - type: derive
      columns:
        aqi: "pm25 * 0.5 + no2 * 0.3"
    - type: aggregate
      group_by: [City]
      aggregations:
        pm25: mean
        aqi: mean

Connectors

Bold = added in v3.0. Add your own connector:

from core.connectors.registry import ConnectorRegistry
ConnectorRegistry.register("my_db", MyDatabaseConnector)

Load Strategies

Transformations

Six declarative transform types run after quality checks, before the DB write:

CLI Reference

openingest run                       # Full pipeline
openingest run --dry-run             # Validate + quality, no DB writes
openingest run --dataset orders      # Single dataset

openingest validate                  # Schema validation
openingest quality                   # Quality scores
openingest report                    # Latest run report
openingest history                   # Run history
openingest dashboard                 # Monitoring dashboard

openingest infer orders.csv          # Infer datasets.yaml from CSV
openingest profile orders.csv        # Profile a CSV

openingest scheduler start --cron @daily
openingest scheduler start --cron "0 6 * * *"

openingest airflow build             # Generate Airflow DAG
openingest docker init               # Generate docker-compose.yml
openingest doctor                    # Check environment
openingest version                   # Show version

Pipeline Flow

Source (CSV · Excel · JSON · Parquet · S3 · Azure · GCS · REST · PostgreSQL · MySQL · MongoDB · SFTP · FTP · Salesforce · HubSpot · Stripe · Google Sheets)
    │
    ▼  Dataset Discovery         core/discovery.py
    ▼  Schema Validation         core/validation.py
    ▼  Data Quality Engine       core/quality.py
    ▼  Transformation Engine     core/transform.py   ← v3.0
    ▼  Ingestion Engine          core/ingestion.py
    ▼  PostgreSQL Staging        auto-created tables
    ▼  Metadata Logger           pipeline_runs · pipeline_dataset_runs
    ▼  Notifications             Slack + Email with retry
    ▼  Airflow DAG               openingest_dynamic_pipeline

Notifications

# configs/pipeline.yaml
notifications:
  slack:
    webhook: ${SLACK_WEBHOOK_URL}
    on: [success, failure]
    retry_count: 3
  email:
    smtp_host: smtp.company.com
    smtp_port: 587
    username: ${EMAIL_USERNAME}
    password: ${EMAIL_PASSWORD}
    to: [data-team@company.com]
    on: [failure]

CI / CD

Links

• Homepage: https://open-ingest.vercel.app
• Repository: https://github.com/ManishKudtarkar/OpenIngest
• Issues: https://github.com/ManishKudtarkar/OpenIngest/issues
• Changelog: https://github.com/ManishKudtarkar/OpenIngest/releases
• PyPI: https://pypi.org/project/openingest/

Milestones

• ✅ v1.0 — Discovery, schema validation, quality engine, incremental loading, Airflow DAG, CLI, CI
• ✅ v2.0 — Excel, JSON, Parquet, S3, Azure Blob, GCS, REST API connectors, plugin registry
• ✅ v2.5 — Built-in scheduler, Slack/email notifications with retry, data lineage engine
• ✅ v3.0 — PostgreSQL, MySQL, MongoDB, SFTP, FTP, Salesforce, HubSpot, Stripe, Google Sheets + transformation engine + 93 tests
• ✅ v3.0.3 — PyPI · pip install openingest
• ✅ v3.0.4 — docs rewrite for PyPI project description
• ✅ v3.0.5 — connector docs updated, 17 connectors table, frontend version sync
• 🔲 v4.0 — Web dashboard, RBAC, multi-environment support, Snowflake/BigQuery connectors

License

MIT — see LICENSE

⭐ If OpenIngest saves you time, star the repo at github.com/ManishKudtarkar/OpenIngest