Load SaaS Data to Snowflake: Key-Pair Auth, SQL API & VARIANT Columns

To load SaaS integration data into Snowflake programmatically, use the Snowflake SQL API with key-pair authentication, and write JSON payloads into VARIANT columns with PARSE_JSON. Truto's Snowflake datastore does the SQL API and JWT work for you, so a sync job can write records with insertRows or executeStatement.

If you are building B2B SaaS, your product's value is directly tied to the data it can ingest. Your customers expect you to pull records from their existing CRMs, HRIS platforms, applicant tracking systems, and accounting tools into your application. You need a reliable way to sync this SaaS data to Snowflake. You do not want to build and maintain 50 custom data pipelines, nor do you want to force customer-facing integrations through an internal ELT tool designed for offline analytics.

Snowflake is one of the most common cloud data warehouses for this kind of workload, so the requirement is simple to state: build the integrations, and get the data into Snowflake safely.

This guide shows how to architect a multi-tenant data sync with Truto's sync job runtime. We cover how to configure Snowflake as a datastore destination, how key-pair authentication works, what the client does on retries and failures, and how to handle schema drift with VARIANT columns.

The Challenge of Syncing SaaS Data to Snowflake at Scale

Bulk data extraction in a B2B SaaS context is fundamentally different from internal data warehousing. When you pull data for internal business intelligence, you control the source, the credentials, and the destination. When you extract data on behalf of your customers, you control none of those things.

The Multi-Tenant Multiplier Effect If you have 1,000 customers and offer 50 integrations, you do not have 50 data pipelines. You have 50,000 unique integration states. Each of those connections has its own OAuth token lifecycle, its own API rate limits, and its own schema customizations.

Many engineering teams attempt to solve this by deploying traditional ETL tools. However, these platforms are built for single-tenant internal data movement, not for syncing data on behalf of each of your customers.

Alternatively, teams build bespoke pipelines using the Snowflake Connector for Python. This leads to large maintenance overhead. You end up writing custom code to handle schema drift, manage OAuth token refreshes, and track incremental sync watermarks across thousands of tenants. This is a challenge we explore deeply in our guide to handling bulk data extraction and ETL workflows through unified APIs.

To build a scalable, multi-tenant sync to Snowflake, you need a declarative pipeline that understands third-party API pagination, manages state, and writes directly to Snowflake's APIs.

Securing Programmatic Access: Snowflake Key-Pair Authentication

Password authentication is a poor fit for API-driven workflows. Passwords in environment variables or connection strings are easy to leak, and the Snowflake SQL API does not accept a plain username and password. Key-pair authentication is the standard choice for a service that writes to Snowflake.

Key-pair authentication uses an RSA public/private key pair to verify identity. You assign the public key to the Snowflake user. The client uses the private key to sign a JSON Web Token (JWT) and sends it in the Authorization header with X-Snowflake-Authorization-Token-Type: KEYPAIR_JWT.

Generate the key pair with openssl. The -nocrypt flag writes an unencrypted PKCS#8 key, which is the format Truto expects:

openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt
openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub

Assign the public key to the Snowflake user. Paste the contents of rsa_key.pub without the -----BEGIN PUBLIC KEY----- and -----END PUBLIC KEY----- lines:

ALTER USER TRUTO_LOADER SET RSA_PUBLIC_KEY='MIIBIjANBgkqh...';

A Snowflake datastore in Truto holds the connection details. You create it with POST /datastore. Put the contents of rsa_key.p8 in private_key, with each line break written as \n:

{
  "type": "snowflake",
  "label": "warehouse-prod",
  "config": {
    "account": "myorg-myaccount",
    "user": "TRUTO_LOADER",
    "private_key": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----",
    "warehouse": "LOAD_WH",
    "database": "PROD_DB",
    "schema": "INTEGRATIONS",
    "role": "TRUTO_LOADER_ROLE"
  }
}

• account is orgname-account_name or a legacy account locator.
• private_key is an unencrypted RSA PEM key, PKCS#8 or PKCS#1. An encrypted key (BEGIN ENCRYPTED PRIVATE KEY) is rejected. Truto encrypts the key at rest and never returns it in API responses.
• warehouse, database, schema and role are case-sensitive.
• Add "validation_methods": ["testConnection"] to the request to run SELECT 1 before the datastore is saved. Validation never writes to a table.

sequenceDiagram
    participant SyncJob as Sync Job Run (V4)
    participant Datastore as Datastore Service
    participant Client as Snowflake Client
    participant Snowflake as Snowflake SQL API
    SyncJob->>Datastore: Get datastore (cached per run)
    Datastore-->>SyncJob: Config with decrypted private key
    SyncJob->>Client: createDatastoreClient() (each write)
    Client->>Client: Sign key-pair JWT (valid 1 hour)
    Client->>Snowflake: POST /api/v2/statements?requestId=...&retry=true
    Snowflake-->>Client: 200 result, or 202 + statementHandle
    Client->>Snowflake: GET /api/v2/statements/{handle} (poll while 202)
    Client-->>SyncJob: Rows, stats, or a failure

When a sync job runs, the runtime fetches the datastore once per run and caches it. The runtime builds a new Snowflake client for each write, that is, each time the destination node runs on a page of records. The client signs a JWT with iss = ACCOUNT.USER.SHA256:<public key fingerprint>. The token is valid for one hour. The client keeps the token for its own statements and signs a new one after 50 minutes. Your application code never handles the private key or the token.

Key takeaway: Do not use static passwords for programmatic warehouse access. Use key-pair authentication, and keep the private key in one encrypted place.

How Truto Sends Statements to the Snowflake SQL API

Modern serverless and API-driven architectures use the Snowflake SQL API instead of a driver with a persistent TCP connection. The core endpoint is POST /api/v2/statements, which takes SQL text and bind variables over HTTP.

Writing over HTTP has a distributed systems problem: network failures. Snowflake can run an INSERT, and the HTTP response can drop before it reaches the caller. A naive retry then runs the statement again and creates duplicate rows.

Snowflake solves this at the request level with the requestId and retry query parameters. If a request is resent with the same requestId and retry=true, Snowflake does not run the statement a second time.

Truto's Snowflake client sends this request body for each statement:

{
  "statement": "INSERT INTO \"RAW_CONTACTS\" (\"REMOTE_ID\", \"NAME\") VALUES (?, ?)",
  "timeout": 60,
  "warehouse": "LOAD_WH",
  "database": "PROD_DB",
  "schema": "INTEGRATIONS",
  "role": "TRUTO_LOADER_ROLE",
  "bindings": {
    "1": { "type": "TEXT", "value": "003xx000004TmiQ" },
    "2": { "type": "TEXT", "value": "Acme Corp" }
  }
}

The client behaves as follows:

• Request-level retries. Each statement gets a new random requestId. The client retries a network error, a 429 or a 5xx with the same requestId and retry=true, so a retry does not run the statement twice.
• Polling. When Snowflake answers 202, the client polls GET /api/v2/statements/{handle}.
• Timeout and cancel. A statement runs with a 60-second timeout, the same as the poll budget. When the polls give up, the client cancels the statement, so the statement cannot commit after the sync job counted it as failed.
• Bindings. Every value is sent as a TEXT binding, and Snowflake casts it to the column type. Objects are sent as JSON text.

This approach is conceptually similar to the challenges faced when streaming data to other warehouses, a topic we cover in our guide on syncing SaaS data to BigQuery streaming inserts and deduplication.

Handling Schema Drift with VARIANT Columns and PARSE_JSON

The most brittle part of any SaaS integration pipeline is the schema mapping layer. Third-party APIs change. Vendors add new fields, deprecate old ones, and allow their users to create unlimited custom fields.

If you map a Salesforce Contact or a Workday Employee only to strict relational columns in Snowflake (first_name VARCHAR, last_name VARCHAR, title VARCHAR), every new field needs a schema change. When a customer adds a custom industry_vertical field in their CRM, a write that includes that field fails because the destination column does not exist.

Snowflake's VARIANT data type addresses this.

A VARIANT column stores semi-structured data such as JSON. A single VARIANT value can hold up to 128 MB. This enables a "schema-on-read" design: you load the raw or lightly mapped JSON record into Snowflake, and parse it at query time.

Create the target table with flat columns for the fields you query often and a VARIANT column for the raw record. Snowflake does not enforce primary keys, so the MERGE below is what keeps (INTEGRATED_ACCOUNT_ID, REMOTE_ID) unique:

CREATE TABLE IF NOT EXISTS PROD_DB.INTEGRATIONS.RAW_CONTACTS (
  INTEGRATED_ACCOUNT_ID VARCHAR NOT NULL,
  REMOTE_ID VARCHAR NOT NULL,
  RAW_DATA VARIANT,
  UPDATED_AT TIMESTAMP_TZ
);

To insert a JSON string into a VARIANT column, wrap the binding in PARSE_JSON(). If you bind a JSON string straight to a VARIANT column, Snowflake stores a string value, not a traversable JSON object. Snowflake does not accept PARSE_JSON inside a VALUES clause, so use INSERT ... SELECT:

-- Insert an API record into a VARIANT column
INSERT INTO RAW_CONTACTS (INTEGRATED_ACCOUNT_ID, REMOTE_ID, RAW_DATA, UPDATED_AT)
SELECT ?, ?, PARSE_JSON(?), ?;

Once the data is stored as a VARIANT, your analytics engineers can query it with path notation:

SELECT
  remote_id,
  raw_data:name::string AS company_name,
  raw_data:custom_fields.industry_vertical::string AS industry,
  updated_at
FROM raw_contacts
WHERE integrated_account_id = 'acc_12345';

With a VARIANT column for the raw record, a new upstream field no longer breaks the write. The sync job fetches the record from the unified API, maps it with JSONata, and writes it to Snowflake.

insertRows vs. executeStatement

Truto's Snowflake datastore has three methods: testConnection, insertRows and executeStatement.

insertRows details that matter in production:

• Identifiers are upper-cased and quoted. A column remote_id becomes "REMOTE_ID", which resolves the same way as an unquoted name. table can be TABLE, SCHEMA.TABLE or DATABASE.SCHEMA.TABLE.
• Objects become JSON text. A VARIANT column then stores a string. For VARIANT, use executeStatement with PARSE_JSON.
• At most 10000 bindings per statement. The client lowers the rows per statement to fit this limit. The default batch_size is 1000 rows.
• Each statement commits on its own. If a later statement fails, the rows from earlier statements stay inserted, and the error says how many rows went in before the failure.

Architecting the Pipeline with Truto Sync Jobs

Building this architecture from scratch means you must paginate through third-party APIs, manage watermarks for incremental syncs, and run batch writes to the Snowflake SQL API.

Truto's sync job runtime (V4) handles this through a resource graph. Each node has a type, and depends_on names its parent. Instead of a bespoke Python script for every integration, you define the sync logic in JSON.

1. State Management and Incremental Pulls

A get_state node reads the last watermark (for example last_synced_at) from the sync job run's state, and an add_context node exposes it to the request nodes. The request then asks the unified API only for records that changed since that time. An update_state node writes the new watermark when the run completes, gated on sync_job_run.status = 'completed'. A run that fails does not advance the watermark. But a failed Snowflake write does not fail the run, so the watermark can move past records that never reached Snowflake (see below).

2. Source API Rate Limits

When a source API returns HTTP 429 during a V4 sync run, the run pauses and resumes by itself. Truto sends a sync_job_run:rate_limited webhook event so you can see the pause. You do not write retry code for this.

3. OAuth Token Refreshes

Syncing large datasets can take a long time, and source OAuth access tokens expire. Truto refreshes the integrated account's token before it expires, and a mutex makes sure that only one refresh runs at a time for the same account. We describe this pattern in our guide to the architecture of reliable token refreshes.

4. Declarative Snowflake Destinations

A JSONata transform node maps the unified API response into the row shape the table expects. Truto does not change the data beyond that JSONata mapping. Here the transform keeps two flat fields and serializes the whole record for the VARIANT column. It writes an explicit null when a record has no updated_at:

{
  "type": "transform",
  "name": "contacts-to-rows",
  "depends_on": "list-crm-contacts",
  "config": {
    "expression": "resources.crm.contacts.{ \"remote_id\": id, \"updated_at\": $exists(updated_at) ? updated_at : null, \"raw_json\": $string($) }"
  }
}

The null matters. If the transform leaves the key out, the placeholder {{payload.records.updated_at}} stays as the literal {{...}} text, and the MERGE fails when it casts that text to a timestamp. An explicit null is bound as SQL NULL. Writing the binding as {{payload.records.updated_at:null}} has the same effect.

The datastore destination iterates the rows with loop_on: "payload.records". Inside each iteration, payload.records is the current row. The MERGE updates a row that already exists and inserts a new one, so a re-run or a replayed page does not create duplicate rows:

{
  "type": "destination",
  "name": "write-contacts-to-snowflake",
  "destination_type": "datastore",
  "method": "executeStatement",
  "loop_on": "payload.records",
  "run_if": "$exists(payload.records) and $type(payload.records) = \"array\" and $count(payload.records) > 0",
  "config": {
    "id": "{{args.snowflake_datastore_id}}",
    "config": {
      "statement": "MERGE INTO RAW_CONTACTS t USING (SELECT ? AS INTEGRATED_ACCOUNT_ID, ? AS REMOTE_ID, PARSE_JSON(?) AS RAW_DATA, ? AS UPDATED_AT) s ON t.INTEGRATED_ACCOUNT_ID = s.INTEGRATED_ACCOUNT_ID AND t.REMOTE_ID = s.REMOTE_ID WHEN MATCHED THEN UPDATE SET t.RAW_DATA = s.RAW_DATA, t.UPDATED_AT = s.UPDATED_AT WHEN NOT MATCHED THEN INSERT (INTEGRATED_ACCOUNT_ID, REMOTE_ID, RAW_DATA, UPDATED_AT) VALUES (s.INTEGRATED_ACCOUNT_ID, s.REMOTE_ID, s.RAW_DATA, s.UPDATED_AT)",
      "bindings": [
        "{{args.integrated_account_id}}",
        "{{payload.records.remote_id}}",
        "{{payload.records.raw_json}}",
        "{{payload.records.updated_at}}"
      ]
    }
  },
  "resources_to_persist": ["contacts-to-rows"]
}

The destination reads two run arguments, args.snowflake_datastore_id and args.integrated_account_id. Declare them in the sync job's args_schema (the job also needs "default_runtime_version": 4). With these formats, Truto checks that both IDs exist in your environment before the run starts:

{
  "args_schema": {
    "integrated_account_id": {
      "type": "string",
      "format": "integrated_account_id",
      "required": true
    },
    "snowflake_datastore_id": {
      "type": "string",
      "format": "datastore_id",
      "required": true
    }
  }
}

Pass the values in args when you create the sync job run:

{
  "sync_job_id": "<sync job id>",
  "args": {
    "integrated_account_id": "<integrated account id>",
    "snowflake_datastore_id": "<datastore id>"
  }
}

bindings is a plain array of values. The client turns each value into a TEXT binding. Always pass record values through bindings, never template them into statement. The runtime replaces placeholders in the whole method config, so a record field templated into statement runs as SQL.

The MERGE runs one statement per record. For high volume with flat columns, insertRows writes up to 1000 rows per statement instead:

{
  "type": "destination",
  "name": "append-contacts-to-snowflake",
  "destination_type": "datastore",
  "method": "insertRows",
  "config": {
    "id": "{{args.snowflake_datastore_id}}",
    "config": {
      "table": "INTEGRATIONS.RAW_CONTACTS_STAGE",
      "rows": "{{payload.records:json}}",
      "batch_size": 1000
    }
  },
  "resources_to_persist": ["contacts-to-rows"]
}

insertRows uses the row keys as column names, so the staging table has one column for each key the transform emits:

CREATE TABLE IF NOT EXISTS PROD_DB.INTEGRATIONS.RAW_CONTACTS_STAGE (
  REMOTE_ID VARCHAR,
  UPDATED_AT TIMESTAMP_TZ,
  RAW_JSON VARCHAR
);

insertRows stores raw_json as text. A Snowflake task can then run PARSE_JSON and a MERGE from the staging table into the final VARIANT table.

Failed Writes Do Not Fail the Run

A failed Snowflake write does not fail the sync job run. The run still ends completed, and the update_state node moves the watermark forward. Truto does not replay the failed records. The only signal is the sync_job_run:record_error webhook event. It carries the error, including how many rows earlier statements inserted, but it does not carry the failed rows. Watch the webhook and backfill by resetting the watermark to a time before the failure.

Strategic Wrap-Up

Syncing multi-tenant SaaS data to Snowflake does not require an internal ETL tool or thousands of lines of brittle Python code.

The Snowflake SQL API, key-pair authentication and VARIANT columns give you a scalable data pipeline that tolerates schema drift. Truto's sync job runtime handles OAuth refreshes, incremental watermarks, and the Snowflake client's JWT signing, request-level retries and statement timeouts. You choose insertRows or a MERGE through executeStatement, depending on whether a re-run must be duplicate-free.

The result is an integration layer that delivers customer data into Snowflake, so your engineering team can focus on core product features instead of babysitting data pipelines.