Adding the E to dbt: Extracting Source Systems With dbt Core and Snowflake

dbt Excels at Transformation, but Extraction Remains Separate

dbt (data build tool) has become a widely used standard for transformations in modern data engineering. It enables engineers to build modular, version-controlled, and tested SQL pipelines, applying established software engineering practices to data work.

However, dbt is designed for the T in traditional ETL or ELT pipelines. It does not inherently cover extraction or loading.

In a conventional ELT setup, dedicated tools such as Fivetran, Airbyte, or custom ingestion scripts retrieve data from databases, SaaS applications, and APIs. While effective, this often adds operational overhead: more tools to operate, more infrastructure to manage, more context switching, and potentially higher costs.

But what if extraction could happen directly in dbt? Rather than running a separate orchestration layer, maintaining a VM, or licensing an additional connector, teams could build SQL-native pipelines that run from the source system through to the data mart.

With Snowflake, this approach is possible through a concise integration pattern.

Snowflake User-Defined Table Functions for Data Extraction

The central capability is that Snowflake supports Python-based User-Defined Table Functions (UDTFs) through Snowpark. These functions can run Python code within Snowflake and return the result as a standard SQL table. Because dbt Core is available natively in Snowflake, there is no requirement to host or deploy dbt separately.

This makes it possible to:

1. Write a Python function that connects to an external source system
2. Register the function as a Snowflake table function
3. Invoke the function from a Snowflake-hosted dbt model using standard SQL

For dbt, this looks like an ordinary table query. In the background, Snowflake connects to the source system, retrieves the requested data, and returns it as rows and columns within one SQL statement.

This creates an end-to-end ELT data pipeline in dbt without introducing a separate technology layer.

Sample Implementation

The following scripts provide an example implementation for Salesforce.

Create a Snowpark Python Table Function

USE DATABASE DB_MAHT_DBT;
USE SCHEMA UTIL;

-- create secrets
CREATE OR REPLACE SECRET DB_MAHT_DBT.UTIL.SALESFORCE_CREDS
  TYPE = GENERIC_STRING
  SECRET_STRING = '{"username": "<username>",
                    "password": "<password>",
                    "security_token": "<security-token>",
                    "instance_url": "<instance-url>"}';

-- create network rule
CREATE OR REPLACE NETWORK RULE DB_MAHT_DBT.UTIL.SALESFORCE_NETWORK_RULE
  MODE = EGRESS
  TYPE = HOST_PORT
  VALUE_LIST = (
    'login.salesforce.com',
    '<instance-url>'
  );

-- create external access integration
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION SALESFORCE_ACCESS_INTEGRATION
  ALLOWED_NETWORK_RULES = (DB_MAHT_DBT.UTIL.SALESFORCE_NETWORK_RULE)
  ALLOWED_AUTHENTICATION_SECRETS = (DB_MAHT_DBT.UTIL.SALESFORCE_CREDS)
  ENABLED = TRUE;

-- create UDTF
CREATE OR REPLACE FUNCTION DB_MAHT_DBT.UTIL.FCT_LOAD_SALESFORCE_TABLE(TARGET_TABLE STRING, FIELD_LIST STRING DEFAULT '', WHERE_CLAUSE STRING DEFAULT '', SOQL_QUERY STRING DEFAULT '')
RETURNS TABLE ("DATA" OBJECT)
LANGUAGE PYTHON
RUNTIME_VERSION = '3.11'
ARTIFACT_REPOSITORY = snowflake.snowpark.pypi_shared_repository
PACKAGES = ('simple-salesforce', 'snowflake-snowpark-python')
EXTERNAL_ACCESS_INTEGRATIONS = (SALESFORCE_ACCESS_INTEGRATION)
SECRETS = ('sf_creds' = DB_MAHT_DBT.UTIL.SALESFORCE_CREDS)
HANDLER = 'pythonDataReader'
AS
$$
import _snowflake
import json
from simple_salesforce import Salesforce

class pythonDataReader:
    def process(self, target_table, field_list, where_clause, soql_query):
        creds = json.loads(_snowflake.get_generic_secret_string('sf_creds'))

        sf = Salesforce(
            username=creds['username'],
            password=creds['password'],
            security_token=creds['security_token'],
            instance_url=creds['instance_url']
        )

        if soql_query and soql_query.strip():
            query = soql_query.strip()
            result = sf.query_all(query)
            records = result['records']
            if records:
                fields = [k for k in records[0].keys() if k != 'attributes']
            else:
                fields = []
        else:
            sf_object = getattr(sf, target_table)

            if field_list and field_list.strip():
                fields = [f.strip() for f in field_list.split(',')]
            else:
                desc = sf_object.describe()
                fields = [f['name'] for f in desc['fields']]

            query = "SELECT " + ", ".join(fields) + " FROM " + target_table
            if where_clause and where_clause.strip():
                query += " WHERE " + where_clause
            result = sf.query_all(query)
            records = result['records']

        for r in records:
            row = {f: str(r.get(f)) if r.get(f) is not None else None for f in fields}
            yield (row,)
$$;

Example UDTF Calls

-- select all columns:
SELECT DATA FROM TABLE(DB_MAHT_DBT.UTIL.FCT_LOAD_SALESFORCE_TABLE('Account'));

-- select specific columns:
SELECT DATA FROM TABLE(DB_MAHT_DBT.UTIL.FCT_LOAD_SALESFORCE_TABLE('Account', 'Id, Name, Industry, BillingCity'));

-- using a filter:
SELECT DATA FROM TABLE(DB_MAHT_DBT.UTIL.FCT_LOAD_SALESFORCE_TABLE('Account', 'Id, Name, Industry', 'Industry = \'Electronics\''));

-- using custom SOQL query:
SELECT DATA FROM TABLE(DB_MAHT_DBT.UTIL.FCT_LOAD_SALESFORCE_TABLE('a', 'b', 'c', 'SELECT Id, Name FROM Account WHERE Id = \'001g500000KiFO3AAN\''));

‍

In this example, the UDTF returns a VARIANT column, Snowflake’s flexible semi-structured data type for JSON-like objects. Subsequent models can extract individual values with Snowflake’s colon notation, for example data:Id::STRING.

Using the Snowpark UDTF in a dbt Model

This is where the components connect. A dbt model can call the UDTF in the same way it would call any other SQL table function, while dbt manages the model execution. The extracted records are materialized as a Snowflake table and can feed directly into the existing transformation pipeline.

{{ config(
    materialized='table'
) }}


SELECT
    data::variant AS data,
    '{{ run_started_at }}'::timestamp(0) AS prj$load_dts
FROM TABLE(DB_MAHT_DBT.UTIL.FCT_LOAD_SALESFORCE_TABLE('Account'))

Conclusion: A Complete ELT Pipeline Within dbt

dbt was created for data transformation, and it provides a strong foundation for that task. Combined with Snowflake Snowpark User-Defined Table Functions, it can also serve as the entry point for extraction.

This pattern allows teams to build an ELT pipeline from the source system to a consumption-ready data mart within a dbt project and Snowflake environment. It avoids the need for an additional extraction tool or separate infrastructure.

Python’s broad ecosystem of libraries can make the approach applicable and scalable to many types of source systems, including:

• Relational databases such as PostgreSQL, Oracle, and SQL Server
• SaaS platforms such as Salesforce, HubSpot, and Jira
• File and collaboration systems such as SharePoint
• REST APIs

For teams that already use dbt and Snowflake, this pattern is worth evaluating before adding another extraction tool to the technology stack.

‍

If you’re also facing data integration challenges and want to create real added value for your company, we’re happy to support you. Contact us for an initial, non-binding consultation to discuss your specific use case.