Snowflake

Read training data from Snowflake with leakage-safe bitemporal mapping and write forecasts back.

rb.sources.snowflake connects deployed Rebase code — a Predictor, step, or workflow — to your Snowflake account. It shares the uniform data-source surface: read, read_bitemporal, and write.

Install

uv pip install "rebase-toolkit[snowflake]"

Ships snowflake-connector-python[pandas], pyarrow, and cryptography (for key-pair auth). Declare it on the image you deploy:

image = rb.Image.python("3.12").uv_pip_install("rebase-toolkit[snowflake]")

Authenticate

Key-pair authentication is recommended for service accounts — no password to rotate through your secret store:

# 1. Generate an RSA key pair (unencrypted PKCS#8 shown; passphrase supported)
openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt
openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub

# 2. Assign the public key to your service user in Snowflake
#    ALTER USER REBASE_SVC SET RSA_PUBLIC_KEY='MIIBIjANBgkq...';

# 3. Store the credentials as a Rebase secret bundle
rebase secret create acme-snowflake \
    SNOWFLAKE_ACCOUNT=xy12345.eu-central-1 \
    SNOWFLAKE_USER=REBASE_SVC \
    SNOWFLAKE_PRIVATE_KEY=-  < rsa_key.p8

Attach the bundle with secrets=[rb.Secret.from_name("acme-snowflake")] — every key becomes an environment variable inside your deployed code. Password auth (SNOWFLAKE_PASSWORD) also works; the connector prefers a key when both are present.

Settings

Setting	Environment variable	Required
`account`	`SNOWFLAKE_ACCOUNT`	✓
`user`	`SNOWFLAKE_USER`	✓
`private_key` / `private_key_path`	`SNOWFLAKE_PRIVATE_KEY` / `SNOWFLAKE_PRIVATE_KEY_PATH`	one auth method
`password`	`SNOWFLAKE_PASSWORD`	one auth method
`passphrase`	`SNOWFLAKE_PRIVATE_KEY_PASSPHRASE`	if the key is encrypted
`role`, `warehouse`, `database`, `schema`	`SNOWFLAKE_ROLE`, …	optional

Any setting can also be passed to the factory directly, or namespaced per connection as REBASE_SOURCE_<CONNECTION>_<FIELD> — see credential precedence.

Read

The Snowflake connector binds parameters with %s (or %(name)s with a dict) and fetches results as Arrow-backed pandas:

import rebase as rb

src = rb.sources.snowflake(connection="acme")
df = src.read(
    "SELECT ts, issued_at, load_mw FROM ANALYTICS.DEMAND WHERE site = %s",
    params=["site-001"],
)

For backtesting, declare knowledge time so emflow can prove there is no leakage:

from rebase.sources import BitemporalSpec

spec = BitemporalSpec(valid_time="ts", knowledge_time="issued_at")
df = src.read_bitemporal(
    "SELECT ts, issued_at, load_mw FROM ANALYTICS.DEMAND WHERE site = %s",
    spec,
    params=["site-001"],
)

Write forecasts back

Writes use Snowflake's native bulk path (write_pandas), creating the table on first write:

src.write(forecast_df, "ANALYTICS.FORECASTS", mode="append")   # default
src.write(forecast_df, "ANALYTICS.FORECASTS", mode="replace")  # overwrite

Reference

Data warehouses overview — shared surface, credentials, BitemporalSpec
Backtesting — what the bitemporal columns enable