Building a Near Real-Time NetSuite to Snowflake Pipeline with Python

Modern finance and operations teams don’t just want reports – they want live analytics. If your source of truth is NetSuite and your warehouse is Snowflake, a smooth pipeline between the two can unlock powerful dashboards for sales, finance, and leadership.

In this blog, we’ll walk through a code-driven approach to move data from NetSuite to Snowflake using Python. We’ll cover:

1. Overall architecture
2. Connecting to NetSuite (REST API / ODBC style)
3. Loading data into Snowflake (Python connector)
4. Simple incremental load logic
5. How to productionize and schedule the pipeline

High-Level Architecture

Typical NetSuite → Snowflake flow:

1. Extract: Pull data from NetSuite (e.g., Transactions, Customers, Items) via REST API / ODBC.
2. Transform (Light): Clean column names, convert dates, normalize booleans, flatten JSON if needed.
3. Load:
   • Write extracted data to a CSV/Parquet file, or
   • Directly stream rows into Snowflake using the Python connector.
4. Incremental Logic: Use last_modified_date or internalId to only pull new/updated records.

import requests
import json
from datetime import datetime, timedelta

ACCOUNT_ID = "YOUR_ACCOUNT_ID"
BASE_URL = f"https://{ACCOUNT_ID.lower()}.suitetalk.api.netsuite.com/services/rest"
CONSUMER_KEY = "YOUR_CONSUMER_KEY"
CONSUMER_SECRET = "YOUR_CONSUMER_SECRET"
TOKEN_ID = "YOUR_TOKEN_ID"
TOKEN_SECRET = "YOUR_TOKEN_SECRET"

# TODO: Implement your OAuth 1.0 signing here or use requests_oauthlib
# from requests_oauthlib import OAuth1

def get_netsuite_transactions(last_modified_from=None):
    """
    Fetch transactions updated after last_modified_from from NetSuite.
    """
    url = f"{BASE_URL}/record/v1/transaction"
    
    params = {}
    if last_modified_from:
        # Example filter param – actual query syntax may vary based on your REST config
        params["lastModifiedDate"] = f"> {last_modified_from}"

    headers = {
        "Content-Type": "application/json",
        "Prefer": "transient",
        # "Authorization": ...  # OAuth header here
    }

    # oauth = OAuth1(CONSUMER_KEY, CONSUMER_SECRET, TOKEN_ID, TOKEN_SECRET)
    # response = requests.get(url, headers=headers, params=params, auth=oauth)

    # For structure reference:
    response = requests.get(url, headers=headers, params=params)  # Replace with auth=oauth in real use

    response.raise_for_status()
    data = response.json()
    # Typically you'll get "items" or "records" in the response:
    return data.get("items", data)