snowflake-rest 0.1.3

Lightweight Snowflake SQL API client for Python. No native connector needed.

snowflake-rest

Lightweight Python client for the Snowflake SQL API. No native connector needed.

Built for serverless environments (AWS Lambda, Cloud Functions) where snowflake-connector-python is too heavy (50+ MB, C dependencies, slow cold starts). Works anywhere Python runs.

Install

pip install snowflake-rest

Dependencies: requests, PyJWT, cryptography

Quick Start

from snowflake_rest import SnowflakeClient

client = SnowflakeClient(
    account="myorg-myaccount",
    user="SVC_USER",
    private_key_path="~/.snowflake/rsa_key.p8",
    database="MY_DB",
    schema="PUBLIC",
    warehouse="COMPUTE_WH",
)

rows = client.query("SELECT * FROM users WHERE status = ?", ["active"])
# [{"ID": 42, "NAME": "Alice", "CREATED_AT": datetime(2026, 1, 15, ...), "BALANCE": Decimal("150.00")}]

Or from environment variables (suited for use-cases like AWS Lambda):

client = SnowflakeClient.from_env()

Features

Querying

# List of dicts (default)
rows = client.query("SELECT * FROM users WHERE age > ?", [25])

# Single row
row = client.query_one("SELECT * FROM users WHERE id = ?", [42])

# Single value
count = client.query_scalar("SELECT COUNT(*) FROM users")

# Single column as flat list
ids = client.query_column("SELECT id FROM users")

# Tuples (lighter than dicts)
rows = client.query("SELECT id, name FROM users", as_tuples=True)
# [(1, "Alice"), (2, "Bob")]

Automatic Type Coercion

Snowflake SQL API returns everything as strings. This library auto-converts using result metadata:

Snowflake Type	Python Type
FIXED (scale=0)	int
FIXED (scale>0)	Decimal
REAL / FLOAT	float
BOOLEAN	bool
TIMESTAMP_*	datetime
DATE	date
TIME	time
VARIANT / OBJECT / ARRAY	dict / list

Disable with type_coercion=False if you want raw strings.

Result Mapping

Map results directly to dataclasses or Pydantic models:

from dataclasses import dataclass

@dataclass
class User:
    ID: int
    NAME: str
    STATUS: str

users = client.query("SELECT * FROM users", row_type=User)
# [User(ID=42, NAME="Alice", STATUS="active"), ...]

Transactions

with client.transaction() as tx:
    tx.set('now_ts', 'CURRENT_TIMESTAMP()')
    tx.execute("UPDATE users SET status=?, updated=$now_ts WHERE id=?", ["inactive", 42])
    tx.execute("INSERT INTO audit (action, ts) VALUES (?, $now_ts)", ["deactivate"])
# Atomic: all succeed or none do

Handles Snowflake's multi-statement binding limitation internally with safe value escaping.

Batch Insert

client.insert_many("users", [
    {"NAME": "Alice", "AGE": 30},
    {"NAME": "Bob", "AGE": 25},
    # ... 10,000 rows
], batch_size=1000)

Async Queries

handle = client.submit("SELECT * FROM huge_table")
print(handle.status())  # "running" | "complete" | "failed"
rows = handle.result()  # blocks until done

Stored Procedures

result = client.call("MY_PROCEDURE", [arg1, arg2])
# Parsed VARIANT result (dict, list, or scalar)

Export (COPY INTO)

export = client.export(
    sql="SELECT * FROM users WHERE status = ?",
    params=["active"],
    stage="@MY_STAGE",
    path="exports/2026/",
)
# ExportResult(rows_unloaded=5000, file_size=102400, path="@MY_STAGE/exports/2026/export_abc.csv.gz")

Streaming (Memory-Efficient)

for row in client.query_stream("SELECT * FROM billion_row_table"):
    process(row)  # one row at a time, partitions fetched lazily

Pandas DataFrames

pip install snowflake-rest[pandas]

df = client.query_df("SELECT * FROM users")
# pandas DataFrame with correct dtypes

Retry on Transient Errors

client = SnowflakeClient(..., retries=3)
# Auto-retries on: connection errors, 429, 502, 503, 504
# Does NOT retry on: syntax errors, permission denied

Query Profiling

def log_query(sql, duration_ms, rows_returned, query_id):
    print(f"[{query_id}] {duration_ms:.0f}ms, {rows_returned} rows")

client = SnowflakeClient(..., on_query=log_query)

Context Manager

with SnowflakeClient.from_env() as client:
    rows = client.query("SELECT 1")
# Session cleanly closed

Serverless Usage

# handler.py — works with AWS Lambda, Google Cloud Functions, Azure Functions, etc.
from snowflake_rest import SnowflakeClient

client = SnowflakeClient.from_env()

def handler(event, context):
    rows = client.query(
        "SELECT * FROM users WHERE status = ?",
        [event["status"]]
    )
    return {"statusCode": 200, "body": rows}

No C extensions, no connector binaries — just pip install snowflake-rest in your deployment package.

CLI

Query Snowflake from your terminal:

export SNOWFLAKE_ACCOUNT=myorg-myaccount
export SNOWFLAKE_USER=SVC_USER
export SNOWFLAKE_PRIVATE_KEY_PATH=~/.snowflake/rsa_key.p8
export SNOWFLAKE_DATABASE=MY_DB
export SNOWFLAKE_WAREHOUSE=COMPUTE_WH

snowflake-rest query "SELECT * FROM users LIMIT 5"
snowflake-rest query "SELECT * FROM users" --format csv > users.csv
snowflake-rest query "SELECT * FROM users" --format json
snowflake-rest query -f report.sql --format csv
snowflake-rest query "SELECT * FROM t WHERE id = ?" --params '[42]'

Configuration

Constructor

SnowflakeClient(
    account="myorg-myaccount",       # Required
    user="SVC_USER",                 # Required
    private_key="-----BEGIN...",     # PEM string/bytes
    private_key_path="/path/key.p8", # OR file path
    private_key_passphrase=b"pass",  # Optional
    database="MY_DB",               # Optional defaults
    schema="PUBLIC",
    warehouse="COMPUTE_WH",
    role="MY_ROLE",
    timeout=45,                      # Query timeout (seconds)
    poll_interval=2.0,               # Async poll interval
    max_poll_time=300.0,             # Max async wait
    type_coercion=True,              # Auto type conversion
    timezone="UTC",                  # Session timezone
    retries=0,                       # Retry on transient errors
    pool_size=10,                    # HTTP connection pool size
    on_query=callback,               # Profiling hook
)

Environment Variables (for from_env())

Variable	Required	Description
SNOWFLAKE_ACCOUNT	Yes	Account identifier
SNOWFLAKE_USER	Yes	Username
SNOWFLAKE_PRIVATE_KEY	One of these	PEM key string
SNOWFLAKE_PRIVATE_KEY_PATH	One of these	Path to .p8 file
SNOWFLAKE_DATABASE	No	Default database
SNOWFLAKE_SCHEMA	No	Default schema
SNOWFLAKE_WAREHOUSE	No	Default warehouse
SNOWFLAKE_ROLE	No	Default role

Error Handling

from snowflake_rest import QueryError, AuthError, TimeoutError

try:
    rows = client.query("SELECT * FROM t")
except QueryError as e:
    print(e.error_code)    # "002003"
    print(e.sql_state)     # "42S02"
    print(e.args[0])       # "Object 'T' does not exist."
except AuthError:
    print("Key/JWT issue")
except TimeoutError:
    print("Query took too long")

Why Not snowflake-connector-python?

	snowflake-rest	snowflake-connector-python
Package size	< 100 KB	50+ MB
Native deps	None	Yes (C extensions)
Lambda-friendly	Yes	Needs layers, slow cold starts
Protocol	REST (SQL API)	Custom (Snowflake protocol)
Auth	JWT keypair only	Multiple methods
Features	Query, transactions, async, export	Full feature set

Use snowflake-connector-python if you need: OAuth/SSO auth, PUT/GET file transfer, or Snowpark. Use snowflake-rest if you want a lightweight client that just works in serverless.

Benchmarks

Each library ran 3 times with the connector executing first to avoid cache bias.

Full results (3 runs each):

	snowflake-rest			snowflake-connector-python
	Run 1	Run 2	Run 3	Run 1	Run 2	Run 3
Import	0.11s	0.08s	0.08s	0.31s	0.29s	0.27s
Connect	1.14s	1.51s	1.17s	19.84s	3.63s	1.38s
Cold query	0.64s	0.73s	0.71s	0.39s	0.60s	0.55s
Warm query	0.63s	0.53s	0.66s	0.36s	0.61s	0.56s
Total	2.52s	2.85s	2.61s	20.90s	5.12s	2.76s

Key takeaway: Query speeds are comparable, but snowflake-rest has a dramatically faster and more consistent connection setup. The official connector's connect time swings from 1.4s to 19.8s depending on warehouse state — in serverless environments where cold starts are frequent, this unpredictability is the main bottleneck.

Install size: 40 MB (with deps) vs 117 MB.

Benchmarks run with warmup=True. See examples repo for benchmark scripts.

License

MIT