Connect dbt models to Mage Pro blocks (Step-by-step tutorial)

Introduction

In Part 1 of our migration series, we demonstrated how moving from dbt Cloud to Mage Pro could be accomplished in under 10 minutes. Now that your dbt project is successfully migrated, it's time to showcase the real power of Mage Pro: orchestrating your dbt models within data pipelines.

This article will show you how to connect your newly migrated dbt models to Mage Pro pipelines, creating a data workflow that fetches data, transforms it through your dbt models, and loads the results into BigQuery. If you haven't completed Part 1 yet, check out our dbt migration blog first to get your dbt project migrated to Mage Pro.

Creating your Mage Pro pipeline and data loader

Now that your dbt project is migrated, let's create your first integrated pipeline. Here's how to get started:

• Navigate to pipelines: From the Mage Pro homepage, hover over the left navigation menu and select "Pipelines."

• Create a new pipeline: Click the green "New pipeline" button. Select the Batch pipeline option and give your pipeline a memorable name like "My dbt Integration Pipeline."

• Add a data loader block: Once in the pipeline editor, click the "Blocks" button, hover over the "Loader" option, and select the appropriate loader type for your data source (API, Database, File, etc.).

For this integration, you can use any data source that fits your project. Your data loader block will handle:

• Fetching data from your chosen source (APIs, databases, files, etc.)

• Performing initial data cleaning and validation

• Running tests to ensure data quality

• Utilizing secrets for secure access to protected resources

Run the data loader: Execute the data loader block to verify it's successfully extracting your data. You should see a preview of your extracted data in the output.

Data loader block example:

import io
import pandas as pd
import requests
from datetime import datetime, timedelta
if 'data_loader' not in globals():
    from mage_ai.data_preparation.decorators import data_loader
if 'test' not in globals():
    from mage_ai.data_preparation.decorators import test

from mage_ai.data_preparation.shared.secrets import get_secret_value

@data_loader
def load_data_from_api(**kwargs):
    url='<https://data.cityofchicago.org/resource/ijzp-q8t2.csv?$order=updated_on> DESC&$limit=10000'
    app_token = get_secret_value('CHICAGO_DATA_API_KEY')

    # Map API columns to what dbt expects (REMOVED x_coordinate and y_coordinate)
    headers = {
        "id": "id",
        "case_number": "case_number",
        "date": "arrest_date",
        "block": "block",
        "iucr": "iucr",
        "primary_type": "primary_type",
        "description": "description",
        "location_description": "location_description",
        "arrest": "arrest",
        "domestic": "domestic",
        "beat": "beat",
        "district": "district",
        "ward": "ward",
        "community_area": "community_area",
        "fbi_code": "fbi_code",
        "year": "arrest_year",
        "updated_on": "updated_on",
        "latitude": "lat",
        "longitude": "lng",
    }

    params = {
        '$$app_token': app_token,
    }

    response = requests.get(url, params=params)

    if response.status_code == 200:
        # Read the CSV data from the response
        raw_crimes_data = pd.read_csv(io.StringIO(response.text), sep=',')

        # **CRITICAL: Drop problematic coordinate columns**
        problematic_columns = ['location', 'x_coordinate', 'y_coordinate']
        for col in problematic_columns:
            if col in raw_crimes_data.columns:
                raw_crimes_data = raw_crimes_data.drop(columns=[col])
                print(f"✅ Dropped problematic '{col}' column")

        # Rename columns to match what dbt expects
        raw_crimes_data.rename(columns=headers, inplace=True)

        # Clean other text columns that might have problematic characters
        text_columns = ['block', 'description', 'location_description']
        for col in text_columns:
            if col in raw_crimes_data.columns:
                raw_crimes_data[col] = raw_crimes_data[col].astype(str)
                # Escape quotes in text fields
                raw_crimes_data[col] = raw_crimes_data[col].str.replace('"', '""', regex=False)
                # Handle NaN values
                raw_crimes_data[col] = raw_crimes_data[col].replace('nan', '')

        # **FIX: Force coordinate columns to be strings to prevent BigQuery type inference issues**
        coordinate_columns = ['x_coordinate', 'y_coordinate']
        for col in coordinate_columns:
            if col in raw_crimes_data.columns:
                # Convert to string and handle NaN values
                raw_crimes_data[col] = raw_crimes_data[col].astype(str)
                raw_crimes_data[col] = raw_crimes_data[col].replace('nan', '')

        # **ALTERNATIVE: Force all numeric-looking columns to be strings**
        numeric_as_string_columns = ['x_coordinate', 'y_coordinate', 'beat', 'district', 'ward', 'community_area', 'arrest_year']
        for col in numeric_as_string_columns:
            if col in raw_crimes_data.columns:
                raw_crimes_data[col] = raw_crimes_data[col].astype(str)
                raw_crimes_data[col] = raw_crimes_data[col].replace('nan', '')

        return raw_crimes_data
    else:
        # Handle error cases
        print(f"Failed to retrieve data. Status code: {response.status_code}")
        return pd.DataFrame()

Integrating your dbt model

With your data loader successfully pulling data, it's time to connect your migrated dbt models.

• Add a dbt block: Click the "Blocks" button again, scroll down to "dbt," and select "Single model or snapshot (from file)."

• Select your dbt model: Choose the dbt model file you want to integrate, such as rpt_crime_counts_by_district.sql, and click "Add to pipelines."

• Connect the blocks: Visually connect your Mage Pro data loader block to your newly added dbt block by dragging a connector between them.

This connection is crucial - it's what allows your dbt models to reference the data from your Mage Pro blocks.

Configuring your source connections in Mage Pro

Once you’ve added your dbt block and connected it to a Mage Pro data loader or transformer block you essentially generate a source seed. The connection will automatically generate a mage_sources.yml file. This file acts as the bridge between your Mage Pro blocks and your dbt models.

After connecting the blocks you should take the following steps:

1. Locate mage_sources.yml: After connecting the blocks, refresh your file editor page and navigate to your dbt folder. You should find the mage_sources.yml file.

2. Update source connections: You'll need to replace the source tables with Mage specific sources:The first source connection should be the name of your dbt source (e.g., mage_your_project_name)The second source connection will be the full name of your Mage Pro data loader block (e.g., Your_Pipeline_Name_Your_Data_Loader_Block_Name)Important: Use single quotes around these names in the YAML file

This configuration ensures your dbt models can correctly reference the data provided by your Mage Pro data loader.

Sample**mage_sources.yml**file:

sources:
- description: Dataframes Mage upstream blocks
  loader: mage
  name: mage_Chicago_Land_Crime_Data
  schema: crimes
  tables:
  - description: Dataframe for block `chicago_crimes_api_data_loader` of the `dbt_youtube_demo`
      mage pipeline.
    identifier: mage_dbt_youtube_demo_chicago_crimes_api_data_loader
    meta:
      block_uuid: chicago_crimes_api_data_loader
      pipeline_uuid: dbt_youtube_demo
    name: dbt_youtube_demo_chicago_crimes_api_data_loader
version: 2

Running your pipeline and verifying data in your warehouse

The block is connected. This is where I would pause before clicking Run. A one-character selector change can be the difference between rebuilding a report with its upstream model in the same run and rebuilding only the report.

For this Chicago crimes example, stg_chicago_crimes is a view. It feeds two report tables: rpt_crime_counts_by_district and rpt_crime_counts_by_type.

Here is how I make the choice:

• I only need the district report, and staging is already current: Select rpt_crime_counts_by_district. dbt should list only that report model.
• I need the district report rebuilt with its upstream model in the same run: Select +rpt_crime_counts_by_district. dbt should include stg_chicago_crimes and the district report.
• I need both report outputs: Select +rpt_crime_counts_by_district +rpt_crime_counts_by_type. dbt should include the shared staging model and both reports.

That leading + is small but important. It tells dbt to bring the upstream dependencies into the same selected graph. Without it, the staging view stays outside this invocation. With it, dbt rebuilds that upstream model before the report.

Pick the Mage block that matches the job

Use Single model or snapshot (from file) when one dbt file is the story you want visible in the pipeline. Mage adds the models referenced by that file as upstream blocks, so the graph stays easy to follow.

Use All models (w/ optional exclusion) when the selector itself is the story, especially when one run needs multiple terminal models. For both Chicago reports, the YAML dbt block contains:

--select +rpt_crime_counts_by_district +rpt_crime_counts_by_type

Before I run it, I make dbt show its work

I would not skip this preview. Open the Mage terminal and ask dbt to list exactly what each selector matches:

dbt ls --select "+rpt_crime_counts_by_district"
dbt ls --select "+rpt_crime_counts_by_district +rpt_crime_counts_by_type"

Read that output like a packing list. The first command should show the staging model and district report. The second should show the staging model and both reports. dbt may list tests attached to those resources too. If the list is empty, or much bigger than you expected, stop there and fix the selector. It is cheaper than debugging the wrong run later.

Build the same graph you just inspected

Once the inventory looks right, use the same selectors with dbt build. This runs the selected models and their tests together:

dbt build --select "+rpt_crime_counts_by_district"
dbt build --select "+rpt_crime_counts_by_district +rpt_crime_counts_by_type"

Mage follows the same idea when these dbt blocks run inside the pipeline: the selected models complete, then Mage runs the tests associated with them. Check the block logs before you call the run finished.

Do not let a green command be the last check

In the schema or dataset configured by your active dbt profile target, verify these relations:

• stg_chicago_crimes as a view
• rpt_crime_counts_by_district as a table
• rpt_crime_counts_by_type as a table when you selected both report outputs

I finish with a row count or a small aggregate against each report table. A green command tells you the job completed. The warehouse result tells you it completed the job you meant to run.

Conclusion

Connecting dbt to Mage is not the hard part. The important part is choosing the smallest honest slice of the graph, previewing what dbt thinks you selected, and checking the relations that actually landed in the warehouse.

Start with one report. Add its upstream dependencies when the pipeline should own the full path. Add a second terminal model only when the same run genuinely needs both outputs. That keeps the pipeline readable and makes a failed run much easier to reason about.

Want to consolidate your data stack under one tool? Schedule a free demo with Mage to get started today!