.PY MODELS

YOUR FIRST PYTHON MODEL

Every Python model is one function called model, taking two arguments and returning a DataFrame:

# models/my_first_python_model.py

def model(dbt, session):
    import pandas as pd

    dbt.config(materialized="federation_python")

    data = pd.DataFrame({
        "id": [1, 2, 3],
        "name": ["Alice", "Bob", "Charlie"],
        "score": [95, 87, 92],
    })

    return data

That's it. Save the file in models/, run dvt run --select my_first_python_model, and a table with columns id, name, and score appears in your target database.

WHY PYTHON MODELS?

SQL is great for data that already lives in a database. Python models are for the data that doesn't:

HOW IT WORKS

When DVT finds a .py file in your models/ folder, it:

Your Python code  →  DuckDB (local)  →  Sling  →  Target Database
                         ↑
              upstream refs/sources
              extracted here first

THE TWO ARGUMENTS: DBT AND SESSION

Every model(dbt, session) function receives two objects:

REFERENCING OTHER MODELS

dbt.ref() reaches any upstream model, SQL or Python. It hands you a pandas DataFrame directly — no .df() call, no relation object in between:

# models/enrich_customers.py

def model(dbt, session):
    import pandas as pd

    dbt.config(materialized="federation_python")

    # Get data from an upstream SQL model
    customers = dbt.ref("stg_customers")     # → pandas DataFrame

    # Transform with pandas
    customers["name_upper"] = customers["name"].str.upper()
    customers["is_vip"] = customers["total_spend"] > 1000

    return customers

When you call dbt.ref("some_model"), DVT extracts that model's table from the database into the local DuckDB cache for you. No connection handling on your side — it just works.

REFERENCING SOURCES

dbt.source() reads from a source defined in sources.yml:

# models/process_orders.py

def model(dbt, session):
    import pandas as pd

    dbt.config(materialized="federation_python")

    # Read from an Oracle source (automatically extracted to DuckDB)
    orders = dbt.source("oracle_ops", "orders")   # → pandas DataFrame

    # Clean and transform
    orders["order_date"] = pd.to_datetime(orders["order_date"])
    orders["amount"] = pd.to_numeric(orders["amount"], errors="coerce")

    return orders

RETURN TYPES

Your model() function must return one of these types:

Recommendation: start with pandas.DataFrame— it's the most widely used and has the best documentation. Reach for polars when a dataset needs more performance.

CONFIGURATION

Python models use dbt.config() the way SQL models use {{ config() }}:

def model(dbt, session):
    dbt.config(
        materialized="federation_python",  # canonical for .py models
        tags=["python", "api"],            # for selection: dvt run --select tag:api
    )
    # ... your code ...

A Python model is a federated model. federation_python is the canonical materialization, and today results land as f_table (incremental Python is chartered, not yet shipped).

Whatever the label, the code runs locally and the returned DataFrame goes through the same federated load pipeline as SQL federated models — target resolution, per-model target=, and bucket targets all behave identically. Warehouse-side materializations (view, ephemeral) don't apply, because the code never runs on the database engine.

USING THE SESSION (DUCKDB)

The session parameter is a DuckDB connection — handy when part of your logic is easier in SQL than in pandas. Register a DataFrame and query it with full DuckDB SQL:

def model(dbt, session):
    import pandas as pd

    dbt.config(materialized="federation_python")

    # Pull upstream data, register it, and go back to SQL
    orders = dbt.ref("stg_orders")            # pandas DataFrame
    session.register("orders", orders)

    return session.sql("""
        SELECT
            category,
            COUNT(*) AS cnt,
            AVG(amount) AS avg_amount
        FROM orders
        GROUP BY category
    """)  # DuckDB relations are a valid return type

API EXTRACTION EXAMPLES

# models/api_users.py

def model(dbt, session):
    import requests
    import pandas as pd

    dbt.config(materialized="federation_python")

    # Fetch data from any REST API
    response = requests.get("https://api.example.com/users")
    response.raise_for_status()

    users = pd.DataFrame(response.json()["data"])
    return users

# models/api_stripe_charges.py

def model(dbt, session):
    import requests
    import pandas as pd
    import os

    dbt.config(materialized="federation_python")

    api_key = os.environ["STRIPE_API_KEY"]

    response = requests.get(
        "https://api.stripe.com/v1/charges",
        headers={"Authorization": f"Bearer {api_key}"},
        params={"limit": 100},
    )
    response.raise_for_status()

    return pd.DataFrame(response.json()["data"])

# models/api_all_contacts.py

def model(dbt, session):
    import requests
    import pandas as pd
    import os

    dbt.config(materialized="federation_python")

    api_key = os.environ["HUBSPOT_API_KEY"]
    all_records = []
    url = "https://api.hubapi.com/crm/v3/contacts"

    while url:
        response = requests.get(
            url,
            headers={"Authorization": f"Bearer {api_key}"},
        ).json()

        all_records.extend(response["results"])

        # Follow pagination
        next_link = response.get("paging", {}).get("next", {}).get("link")
        url = next_link  # None when no more pages

    return pd.DataFrame(all_records)

# models/enriched_orders.py

def model(dbt, session):
    import requests
    import pandas as pd

    dbt.config(materialized="federation_python")

    # Get orders from your database (via dbt.ref)
    orders = dbt.ref("stg_orders")

    # Get exchange rates from an API
    response = requests.get("https://api.exchangerate.host/latest")
    rates = response.json()["rates"]

    # Enrich: convert amounts to USD
    orders["usd_amount"] = orders.apply(
        lambda row: row["amount"] / rates.get(row["currency"], 1),
        axis=1,
    )

    return orders

ML & AI EXAMPLES

# models/churn_predictions.py

def model(dbt, session):
    import pandas as pd
    import joblib

    dbt.config(materialized="federation_python")

    # Load customers from upstream model
    customers = dbt.ref("dim_customers")

    # Load pre-trained model
    clf = joblib.load("models/churn_model.pkl")

    # Score each customer
    features = customers[["tenure", "monthly_spend", "support_tickets"]]
    customers["churn_probability"] = clf.predict_proba(features)[:, 1]
    customers["churn_risk"] = pd.cut(
        customers["churn_probability"],
        bins=[0, 0.3, 0.7, 1.0],
        labels=["low", "medium", "high"],
    )

    return customers

# models/sentiment_analysis.py

def model(dbt, session):
    import pandas as pd
    from openai import OpenAI
    import os

    dbt.config(materialized="federation_python")

    client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
    reviews = dbt.ref("raw_reviews")

    def classify(text):
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": f"Classify sentiment (positive/negative/neutral): {text}"}],
        )
        return response.choices[0].message.content.strip().lower()

    reviews["sentiment"] = reviews["review_text"].apply(classify)
    return reviews

INSTALLING PYTHON PACKAGES

Python models can use any package installed in your environment. If one is missing, DVT shows a clear error:

DVT062: Python model failed to import: No module named 'requests'

Install packages with pip install or uv add before running your models (dvt sync covers pandas). Common companions:

DAG POSITIONING

Python models fit anywhere in your DAG. Here are the three common patterns:

# models/api_products.py
def model(dbt, session):
    import requests, pandas as pd
    dbt.config(materialized="federation_python")
    return pd.DataFrame(requests.get("https://api.example.com/products").json())

# models/product_report.sql  (downstream, refs the Python model)
# SELECT * FROM {{ ref('api_products') }} WHERE is_active = true

# models/enriched_customers.py
def model(dbt, session):
    import pandas as pd
    dbt.config(materialized="federation_python")
    customers = dbt.ref("stg_customers")         # upstream SQL model
    customers["segment"] = customers["spend"].apply(
        lambda x: "high" if x > 1000 else "low"
    )
    return customers

# models/customer_dashboard.sql  (downstream)
# SELECT segment, COUNT(*) FROM {{ ref('enriched_customers') }} GROUP BY 1

# models/executive_report.py
def model(dbt, session):
    import pandas as pd
    dbt.config(materialized="federation_python")
    revenue = dbt.ref("fct_revenue")
    customers = dbt.ref("dim_customers")
    summary = revenue.groupby("month").agg(total=("amount", "sum")).reset_index()
    summary["customer_count"] = len(customers)
    return summary

HOW DVT PYTHON MODELS DIFFER FROM DBT'S

dbt also supports Python models, but they work very differently — dbt's run on the warehouse, DVT's run on your machine:

In short: dbt Python models are limited to running on expensive cloud warehouses that support server-side Python. DVT Python models run locally with full Python capabilities and work with anytarget database — including PostgreSQL, Oracle, SQL Server, and all others that dbt Python models don't support.

RULES AND CONSTRAINTS

ERROR CODES

QUICK CHECKLIST

Before running a Python model, make sure: