snowflake-rest 0.1.4

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

snowflake-rest

PyPI | GitHub | Issues

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

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

Table of Contents

• Install
• Quick Start
• Features
• Serverless Usage
• CLI
• Configuration
• Error Handling
• Comparison with snowflake-connector-python
• Benchmarks
• License

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 serverless):

client = SnowflakeClient.from_env()

Features

Queries

# 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")]

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

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

Pandas

pip install snowflake-rest[pandas]

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

Retries

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)

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

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

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")

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

Using the Snowflake SQL API directly requires:

1. Load RSA private key, compute SHA-256 fingerprint of the public key DER
2. Generate JWT (iss = ACCOUNT.USER.FINGERPRINT, RS256, 1hr expiry)
3. Cache and refresh tokens before expiry
4. POST to /api/v2/statements with Bearer token + KEYPAIR_JWT header
5. Handle 202 async promotion — poll statementStatusUrl until 200
6. Handle retries on 429/502/503/504
7. For transactions: bindings only apply to the first statement,
   so manually escape and inline values for all subsequent statements
8. Parse paginated results across multiple partitions
9. Coerce all values from strings (the API returns everything as strings)

That typically means maintaining a custom client of 800–1,000 lines, plus boilerplate at every call site:

# Snowflake SQL API — transaction with manual bindings
statements = [
    {"sql": "SET now_ts = CURRENT_TIMESTAMP()", "bindings": {}},
    {
        "sql": 'UPDATE users SET "STATUS"=?, "DISPOSITION"=?, '
               '"UPDATED_TS"=$now_ts WHERE "ID"=?',
        "bindings": {
            "1": {"type": "TEXT", "value": status},
            "2": {"type": "TEXT", "value": disposition},
            "3": {"type": "FIXED", "value": str(uid)},
        },
    },
    {
        "sql": 'INSERT INTO audit_log ("USER_ID", "ACTION", "DETAIL", "TS") '
               "VALUES (?, ?, ?, $now_ts)",
        "bindings": {
            "1": {"type": "FIXED", "value": str(uid)},
            "2": {"type": "TEXT", "value": "status_change"},
            "3": {"type": "TEXT", "value": f"{status}/{disposition}"},
        },
    },
]
result = client.execute_transaction(statements)
# + the 800–1,000 lines behind execute_transaction:
#   JWT fingerprinting, token caching, async polling,
#   value escaping, pagination, retry logic, etc.

With snowflake-rest, all of the above is handled internally:

with client.transaction() as tx:
    tx.set("now_ts", "CURRENT_TIMESTAMP()")
    tx.execute(
        'UPDATE users SET "STATUS"=?, "DISPOSITION"=?, "UPDATED_TS"=$now_ts WHERE "ID"=?',
        [status, disposition, uid],
    )
    tx.execute(
        'INSERT INTO audit_log ("USER_ID", "ACTION", "DETAIL", "TS") '
        "VALUES (?, ?, ?, $now_ts)",
        [uid, "status_change", f"{status}/{disposition}"],
    )

Benchmarks

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

	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

Query speeds are comparable, but snowflake-rest has a 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