snowflake-rest 0.1.0

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

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
    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.

License

MIT