tha-snowflake-runner

A Tabular Helper API library that wraps snowflake-connector-python with typed connection management, multi-format profile support, and a normalized query return shape.

Install

pip install tha-snowflake-runner

Quick start

from tha_snowflake_runner import ThaSnowflake

sf = ThaSnowflake(role="ANALYST", warehouse="COMPUTE_WH", database="PROD", schema="PUBLIC")

# inline SQL
result = sf.query("SELECT id, name FROM users WHERE active = %s", params=(True,))
# {"rows": [{"ID": "u1", "NAME": "Alice"}], "rowcount": 1, "status": None}

# or load SQL from a file
result = sf.query(file="queries/users.sql", params=(True,))

# prints "Getting data from Snowflake" as a tqdm progress bar while fetching (on by default)
result = sf.query("SELECT * FROM users")

# desc= prefixes the bar with a step label: "Step 1 of 7: Getting data from Snowflake"
result = sf.query("SELECT * FROM users", desc="Step 1 of 7")

Connection modes

Mode 1 — native connections.toml (default)

Delegates profile lookup to the connector, which reads ~/.snowflake/connections.toml (or the path in SNOWFLAKE_HOME).

sf = ThaSnowflake(role="ANALYST", warehouse="WH")

# Named profile
sf = ThaSnowflake(connection_name="prod", role="ANALYST", warehouse="WH")

Mode 2 — custom connections file

Pass any .toml, .ini/.cfg, or .json file. Both flat ([profile]) and nested ([connections.profile]) TOML styles are supported.

sf = ThaSnowflake(
    connections_file="~/my_connections.toml",
    connection_name="prod",
    role="ANALYST",
    warehouse="WH",
)

# List available profile names from the file
names = sf.list_profiles()  # ["default", "prod", "dev"]

# Module-level convenience — no ThaSnowflake instance needed
from tha_snowflake_runner import list_profiles
names = list_profiles("~/my_connections.toml")

Mode 3 — inline (no file)

Supply all connection parameters directly. account being set triggers inline mode.

# Okta SSO — opens a browser window
sf = ThaSnowflake(account="myorg", user="me@example.com", authenticator="externalbrowser")

# Password (service accounts)
sf = ThaSnowflake(account="myorg", user="svc@example.com", password="secret")

# Key-pair — from a file
sf = ThaSnowflake(
    account="myorg",
    user="svc@example.com",
    private_key_file="~/keys/rsa_key.p8",
    private_key_passphrase="mypass",
)

# Key-pair — from a PEM string or DER bytes (e.g. pulled from a secrets manager)
# private_key_file and private_key are mutually exclusive.
sf = ThaSnowflake(
    account="myorg",
    user="svc@example.com",
    private_key=os.environ["SNOWFLAKE_PRIVATE_KEY"],  # PEM text
    private_key_passphrase="mypass",  # only used if the PEM is encrypted; ignored for DER
)

# OAuth token
sf = ThaSnowflake(account="myorg", authenticator="oauth", token="...")

Pass quiet_connect=True to suppress connector stdout (e.g. the externalbrowser browser-open message).

Query return shape

Every query() call returns a normalized dict:

result = sf.query("SELECT * FROM orders WHERE id = %s", params=("o1",))
# {
#   "rows": [{"ID": "o1", "TOTAL": 99.0}],
#   "rowcount": 1,
#   "status": None,        # None on success, error string on Snowflake query failure
# }

sf.rows always holds the result of the most recent query (thread-local).

A tqdm progress bar labeled "Getting data from Snowflake" prints while rows are fetched — on by default. Pass desc="..." to prefix it with a step label (e.g. desc="Step 1 of 7" → "Step 1 of 7: Getting data from Snowflake"). Pass show_progress=False to suppress it entirely.

Session — multiple queries on one connection

with sf.session(role="ANALYST", warehouse="WH") as sess:
    users = sess.query("SELECT * FROM users")
    orders = sess.query("SELECT * FROM orders")
    # one connection opened, two queries run, connection closed on exit

For inline use without a with block, use open_session() — caller is responsible for closing:

sess = sf.open_session(role="ANALYST", warehouse="WH")
result = sess.query("SELECT * FROM users")
sess.close()

Pass accumulate=True to append rows across queries into sess.rows:

with sf.session(accumulate=True) as sess:
    sess.query("SELECT * FROM users WHERE region = 'US'")
    sess.query("SELECT * FROM users WHERE region = 'EU'")
    all_users = sess.rows  # combined result from both queries

The return value of each sess.query() is always the current query's result only — sess.rows is the accumulation point.

Threading

sf.rows is thread-local — each thread sees only its own query results. For concurrent workloads, open a Session inside each worker:

from concurrent.futures import ThreadPoolExecutor

def worker(sql):
    with sf.session() as sess:
        return sess.query(sql)

with ThreadPoolExecutor(max_workers=4) as pool:
    results = list(pool.map(worker, queries))

Retry

sf = ThaSnowflake(
    account="myorg",
    retry_connect=3,         # up to 3 retries (4 total attempts)
    retry_on=(OSError,),     # only retry on these exception types
    retry_delay=2.0,         # seconds between attempts
    status_cb=print,         # called with a message on each failed attempt
)

API

ThaSnowflake(*, ...)

Methods

Session

Obtain via sf.session(). Not thread-safe — one Session per thread.

list_profiles(path) — module-level

from tha_snowflake_runner import list_profiles

names = list_profiles("~/connections.toml")  # ["default", "prod", "dev"]

Supports the same formats as Mode 2: .toml, .ini, .cfg, .json.

Scope

tha-snowflake-runner is read-only by design — query() and stream() only. Write operations (INSERT, UPDATE, DELETE, MERGE, DDL) are intentionally not supported; use the raw connector directly for those.

Alternatives

• snowflake-connector-python — the official Snowflake connector; tha-snowflake-runner is a thin typed convenience layer on top of it
• snowflake-sqlalchemy — SQLAlchemy dialect for Snowflake with ORM support
• snowpark-python — Snowflake's DataFrame/ML API for in-warehouse computation

tha-snowflake-runner is intentionally narrow: no ORM, no Snowpark, no async — just a thin typed wrapper for common query patterns and connection-file management.

License

MIT