Snowflake¶

Use SnowflakeConfig to run SQL, stream Arrow results, and load warehouse tables through snowflake-connector-python. The adapter is synchronous and owns a bounded connection pool. Install sqlspec[snowflake] to include the connector and PyArrow.

Installation and Quickstart¶

uv add "sqlspec[snowflake]"

pip install "sqlspec[snowflake]"

pdm add "sqlspec[snowflake]"

poetry add "sqlspec[snowflake]"

Configure a connector profile named prod in ~/.snowflake/connections.toml, then run:

from sqlspec import SQLSpec
from sqlspec.adapters.snowflake import SnowflakeConfig

spec = SQLSpec()
config = spec.add_config(
    SnowflakeConfig(connection_config={"connection_name": "prod"})
)
try:
    with config.provide_session() as session:
        print(session.select_value("SELECT :value::INTEGER", {"value": 42}))
finally:
    config.close_pool()

Output: 42. Session contexts release their connection; close the pool when the application stops. Commit DML explicitly or use session.transaction(). execute_many() follows the connection's transaction settings; wrap it in session.transaction() when a batch must be atomic. DDL commits independently, and savepoint methods raise NotImplementedError.

Choosing a Snowflake Adapter¶

Authentication and Session Settings¶

Read credentials from your environment or secret manager.

from sqlspec.adapters.snowflake import SnowflakeConfig
config = SnowflakeConfig(connection_config={
    "account": "org-account", "user": "warehouse_service",
    "auth_type": "workload_identity", "workload_identity_provider": "AWS",
})

import os
from sqlspec.adapters.snowflake import SnowflakeConfig
config = SnowflakeConfig(connection_config={
    "account": "org-account", "user": "warehouse_service", "auth_type": "pat",
    "auth_token_provider": lambda: os.environ["SNOWFLAKE_TOKEN"],
    "warehouse": "ANALYTICS", "database": "REPORTING", "schema": "PUBLIC",
})

from pathlib import Path
from sqlspec.adapters.snowflake import SnowflakeConfig
config = SnowflakeConfig(connection_config={
    "account": "org-account", "user": "warehouse_service", "auth_type": "jwt",
    "private_key": Path("service-key.p8"),
})

from sqlspec.adapters.snowflake import SnowflakeConfig
config = SnowflakeConfig(connection_config={
    "connection_name": "prod", "timezone": "UTC", "query_tag": "sales-report",
    "session_parameters": {"QUOTED_IDENTIFIERS_IGNORE_CASE": False},
})

For local Embucket testing, set host="localhost", port=3000, and protocol="http" with local credentials. Embucket covers local protocol tests; cloud authentication and stage uploads require Snowflake.

Connection and Pool Configuration¶

Pass connection and pool keys in connection_config, and feature flags in driver_features. The pool defaults to zero initial connections and five maximum connections. Acquisition waits up to 30 seconds, and a pool inherited through a process fork must be recreated.

Driver Features¶

Arrow Streaming and Bulk Loads¶

select_to_arrow() returns an ArrowResult; get_data() exposes its Arrow table. select_stream() reads batches lazily, and readers close their cursor on exhaustion, early close, or session close. Reads use fetch_arrow_batches() without pandas, preserve empty schemas, and fall back to rows for unsupported commands. Materialization promotes compatible Arrow schemas; streams cast later batches to the first schema. Uppercase column names become lowercase by default; mixed-case names remain unchanged.

load_from_arrow(table, source) uses bound batches below the stage threshold, which defaults to 1000 rows for config-managed sessions. Standalone loads own one DML transaction; existing caller transactions stay open. With overwrite=True, existing rows are deleted inside that transaction, so a failed load keeps them. if_not_exists=True table creation retains Snowflake DDL commit behavior. Larger loads write temporary Parquet, upload it with a user-stage PUT, run COPY INTO inside the load transaction, then remove staged files. Use stage_table=True with conflict_keys for staged MERGE; load_from_storage() accepts the same conflict options. See Arrow and storage.

SQL, Query Jobs, and Extension Stores¶

Write :name, :1, %s, or %(name)s parameters; the default wire format is ?. Snowflake stages (@stage, @%table, @~), $1 stage/pipe columns, session variables, and // comments remain native SQL. For VARIANT values, use INSERT INTO events (payload) SELECT PARSE_JSON(:payload) and bind JSON text. MERGE INTO ... USING (SELECT PARSE_JSON(:payload) AS payload) also accepts bound JSON.

execute_script() binds supplied values statically, splits SQL with SQLSpec's Snowflake splitter, and executes each statement through its own managed cursor. Scripting blocks, comments, and dollar-quoted bodies stay intact. Transaction ownership comes from CURRENT_TRANSACTION() on the server. execute_async_query() submits a warehouse query and returns its query ID; fetch_async_query_result(query_id) waits for its result. These methods are synchronous Python calls. Async frameworks use sqlspec.utils.sync_tools.async_() to bridge synchronous work.

Litestar uses SnowflakeStore for binary sessions. Events use SnowflakeEventQueueStore with VARIANT payloads and the poll_queue backend by default. Google ADK uses SnowflakeADKStore and SnowflakeADKMemoryStore; memory recall uses ILIKE, and vector embeddings are unsupported. Built-in schema reconciliation creates extension tables without secondary indexes.

Tenant Isolation¶

Set driver_features={"tenant_strategy": "rls"} (or True, "default", "session_attributes") to bind each session to a tenant for row access policies. Each lease opens its own connection, sets the tenant as an immutable session attribute with CALL SET_SYS_CONTEXT('SNOWFLAKE$SESSION_ATTRIBUTES', ...), reads it back with SYS_CONTEXT, and closes the connection when the session ends. Snowflake cannot overwrite or clear an immutable session attribute, so tenant sessions are never returned to the pool. Setting your own session attributes needs no extra privilege.

CREATE ROW ACCESS POLICY tenant_rows AS (tenant_col VARCHAR) RETURNS BOOLEAN ->
  tenant_col = SYS_CONTEXT('SNOWFLAKE$SESSION_ATTRIBUTES', 'tenant_id');
ALTER TABLE orders ADD ROW ACCESS POLICY tenant_rows ON (tenant_id);

from sqlspec.adapters.snowflake import SnowflakeConfig

config = SnowflakeConfig(
    connection_config={
        "account": "my_org-my_account",
        "user": "APP_SERVICE",
        "warehouse": "APP_WH",
    },
    driver_features={"tenant_strategy": "rls"},
)

with config.provide_session(tenant="acme") as session:
    session.select("SELECT order_id FROM orders")

SYS_CONTEXT returns NULL for an unset attribute, so the policy above hides every row from sessions without a tenant. Attribute names are case-sensitive. Use SnowflakeSessionAttributeStrategy(tenant_attribute="tenant_id", variable_keys={"region": "tenant_region"}) to choose the attribute names and map TenantContextPayload variables to more attributes. create_connection() is rejected on tenant configurations; use provide_session() or provide_connection(). ADBC Snowflake configurations do not support tenant_strategy. See Multi-Tenancy.

Type Mappings and Related Reference¶

Catalog key reflection defaults to PUBLIC; qualified quoted table names retain their exact identity. Strict schema reconciliation recognizes VARIANT as JSON while rejecting unrelated STRUCT/OBJECT drift. See the adapter API and catalog reference.