dbt-bridge 0.1.1

A dbt-native Reverse ETL tool powered by dlt to move data between databases and APIs.

dbt-bridge

A dbt-native Reverse ETL and Cross-Database Movement tool powered by dlt.

dbt-bridge allows you to move data between databases, APIs, and warehouses directly within your dbt Python models. It leverages dlt (Data Load Tool) for robust, schema-aware data loading.

Features

• Cross-Database Movement: Move data from Postgres to Snowflake, S3 to BigQuery, etc.
• Reverse ETL: Push your modeled dbt data to external destinations (Salesforce, HubSpot, Postgres, etc.).
• API Ingestion: Fetch data from APIs, transform it with Pandas in-memory, and load it to your warehouse.
• The "Bridge Pattern": Extract -> Transform (Local) -> Load (Remote).
• Lineage: Registers "Ghost Sources" in dbt so your lineage graph remains complete.

Installation

Install dbt-bridge in your dbt environment. You must include the extras for your specific source/destination.

# Example: Snowflake destination, Postgres source
pip install "dbt-bridge[snowflake,postgres]"

# Example: All supported extras
pip install "dbt-bridge[all]"

Supported Extras

• Warehouses: snowflake, bigquery, redshift, databricks, synapse, fabric
• Databases: postgres, mssql, duckdb, trino, athena
• Storage: s3, gcs, azure, filesystem

Usage Patterns

1. Database-to-Database Transfer

Move a table from a source database (e.g., Postgres) to your destination (e.g., Snowflake).

import dbt_bridge
import dlt
from dlt.sources.sql_database import sql_database

def model(dbt, session):
    dbt.config(materialized='table', packages=['dbt-bridge', 'dlt', 'psycopg2-binary'])

    # 1. Define Source (e.g., Postgres)
    # Credentials loaded from secrets.toml
    source = sql_database(schema="public", table_names=["users"])
    
    # 2. Explicit Lineage (Important!)
    dbt.source("postgres_prod", "users")

    # 3. Define Destination (e.g., Snowflake)
    destination = dlt.destinations.snowflake()

    # 4. Transfer
    return dbt_bridge.transfer(
        dbt=dbt,
        source_data=source,
        target_destination=destination,
        dataset_name="raw_postgres",
        table_name="users_synced"
    )

2. API-to-Database (with Transformation)

Fetch data from an API, transform it with Pandas, and load it.

import dbt_bridge
import pandas as pd
from dlt.sources.helpers.rest_client import RESTClient

def model(dbt, session):
    dbt.config(materialized='table', packages=['dbt-bridge', 'dlt', 'pandas'])

    # 1. Fetch Data
    client = RESTClient(base_url="https://api.example.com")
    data_generator = client.paginate("/users")
    
    # 2. Convert to DataFrame & Transform
    # Use the helper to flatten/convert dlt generators to Pandas
    df = dbt_bridge.api_to_df(data_generator)
    df['email'] = df['email'].str.lower()

    # 3. Load
    destination = dlt.destinations.snowflake()
    
    return dbt_bridge.transfer(
        dbt=dbt,
        source_data=df,
        target_destination=destination,
        dataset_name="raw_api",
        table_name="users"
    )

3. The "Bridge Pattern" (Cross-Database Transformation)

Extract from Source A -> Model Locally (DuckDB) -> Push to Destination B.

1. Ingest Model: Python model fetches data from Source A and returns a DataFrame. dbt saves this as a local table.
2. Transform Model: SQL model reads the local table and applies business logic.
3. Push Model: Python model reads the final SQL model and pushes it to Destination B.

Push Model Example:

import dbt_bridge
import dlt

def model(dbt, session):
    dbt.config(materialized='table')

    # 1. Read Final Model
    # Convert to Arrow/Pandas for dlt compatibility
    final_df = dbt.ref("final_users").arrow() 

    # 2. Push to Destination
    destination = dlt.destinations.snowflake()
    
    return dbt_bridge.transfer(
        dbt=dbt,
        source_data=final_df,
        target_destination=destination,
        dataset_name="analytics_prod",
        table_name="final_report"
    )

Configuration

dlt uses a .dlt/secrets.toml file (or environment variables) for credentials. Place this in your dbt project root.

[destination.snowflake.credentials]
database = "ANALYTICS"
password = "password"
username = "user"
host = "account_id" # Do not include .snowflakecomputing.com
warehouse = "COMPUTE_WH"

[sources.sql_database.credentials]
drivername = "postgresql"
database = "db_name"
password = "password"
username = "user"
host = "host"
port = 5432

License

MIT