Data Warehouses

Use the data-warehouse sources when your data already lives in Snowflake, Databricks, BigQuery, or Microsoft Fabric. rebase.sources gives every warehouse the same read/write surface and a leakage-safe bitemporal mapping, so a Predictor can train on warehouse data and be backtested honestly, then write forecasts back.

The connector runs inside your deployed function, step, or model — the same place your fit/predict code runs. Rebase never sees your warehouse credentials or data: it only injects the credentials you configure as environment variables (via secrets=), and your code queries the warehouse directly.

Install

Each backend has its own optional extra so your image only carries the driver it uses:

uv pip install "rebase-toolkit[snowflake]"
uv pip install "rebase-toolkit[databricks]"
uv pip install "rebase-toolkit[bigquery]"
uv pip install "rebase-toolkit[fabric]"

Install all four with rebase-toolkit[sources]. Declare the extra on the image you deploy:

import rebase as rb

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

Credentials

Credentials resolve in this order, highest priority first:

1. explicit keyword arguments to the factory,
2. REBASE_SOURCE_<CONNECTION>_<FIELD> environment variables when you pass connection="<name>",
3. provider-standard environment variables (e.g. SNOWFLAKE_ACCOUNT).

Supply credentials as a secret: a named bundle of environment variables, created once and attached to any function, model, or asgi_app. Create the bundle with the CLI — values go straight to Secret Manager and are never stored by Rebase:

$ rebase secret create acme-snowflake \
    SNOWFLAKE_ACCOUNT=xy12345.eu-central-1 \
    SNOWFLAKE_USER=REBASE_SVC \
    SNOWFLAKE_PRIVATE_KEY=-  < rsa_key.p8
Created secret acme-snowflake with keys: SNOWFLAKE_ACCOUNT, SNOWFLAKE_PRIVATE_KEY, SNOWFLAKE_USER
Use it with: secrets=[rebase.Secret.from_name("acme-snowflake")]

KEY=- reads that value from stdin; --from-dotenv .env.acme imports a dotenv file; --force overwrites an existing bundle. Inspect with rebase secret list (names and key names only — never values) and remove with rebase secret delete.

Attach the bundle in code — every key becomes an environment variable at run time:

class DemandForecast(rb.Predictor):
    name = "demand-forecast"
    image = rb.Image.python("3.12").uv_pip_install("rebase-toolkit[snowflake]")
    secrets = [rb.Secret.from_name("acme-snowflake")]

    def fit(self, train) -> None: ...
    def predict(self, obs) -> dict: ...

secrets= is accepted on functions, models (including Predictor), and asgi_apps, and takes a list of secrets — attach several bundles and their keys merge (later bundles win). Besides from_name, you can build secrets from local values at deploy time:

rb.Secret.from_dict({"SNOWFLAKE_ACCOUNT": "xy12345"}, name="acme-snowflake")
rb.Secret.from_dotenv(".env.acme", name="acme-snowflake")

Both create (or update) the workspace bundle when you deploy, then attach it — handy in CI. Prefer from_name + rebase secret create for credentials shared across a team.

For non-sensitive config there is also env={...} with literal values, but env= values are stored in plaintext on the version record — never put real credentials there.

Read

Every source exposes read(query, *, params=None) returning a pandas DataFrame:

src = rb.sources.bigquery(connection="acme")
df = src.read("SELECT ts, load_mw FROM `acme.energy.demand` WHERE site = @site",
              params={"site": "site-001"})

Leakage-safe reads with read_bitemporal

The number-one failure mode of an energy-forecasting backtest is leakage — a model reading data that was not knowable at forecast time. A raw warehouse table usually has only a target timestamp, so you declare how to derive the knowledge time (when each row became available) with a BitemporalSpec:

from datetime import timedelta

from rebase.sources import BitemporalSpec

# Best: the table records when each row became knowable.
spec = BitemporalSpec(valid_time="ts", knowledge_time="issued_at")

# Or: availability lags the target time by a fixed, known delay.
spec = BitemporalSpec(valid_time="ts", knowledge_delay=timedelta(hours=1))

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

read_bitemporal returns a frame with canonical valid_time and knowledge_time columns, ready to feed to emflow for a backtest where leakage is structurally impossible.

Write forecasts back

Every source also exposes write(df, table, *, mode="append") (mode="replace" truncates first), so a deployed model can push forecasts back to the warehouse:

src = rb.sources.snowflake(connection="acme")
src.write(forecast_df, "ANALYTICS.FORECASTS", mode="append")

BigQuery and Snowflake use native bulk load paths. Databricks and Fabric writes use batched INSERT statements suited to modest result sets such as forecasts; for large bulk loads, stage Parquet and use COPY INTO (Databricks) or the native bulk loader (Fabric).

The energy data model (BigQuery)

For energy time series, the BigQuery source goes beyond generic SQL: it stores data in the canonical layout of rebase's open data model (energydatamodel / timedatamodel) — the same schema rebase uses internally — so warehouse data is structurally interchangeable with platform data.

The layout is an append-only series_values table with three time axes plus a series catalog:

A series is keyed the energydb way — (path, data_type, name), e.g. ("portfolio/site-1/t01", "forecast", "electricity.supply") — with unit, resolution, FLAT/OVERLAPPING type, and retention in the catalog. series_id is derived deterministically from the key, so registration is idempotent.

src = rb.sources.bigquery(connection="acme")
src.ensure_energy_schema("energy")   # creates series_values + series (idempotent)
src.register_series("energy", path="portfolio/site-1/t01", data_type="forecast",
                    name="electricity.supply", unit="MW", timeseries_type="OVERLAPPING")

# write a forecast issue: VERSIONED shape = knowledge_time + valid_time + value
src.write_series("energy", forecast_frame, path="portfolio/site-1/t01",
                 data_type="forecast", name="electricity.supply")

Reads replicate timedb's point-in-time semantics — the properties that make backtests honest:

key = ("portfolio/site-1/t01", "forecast", "electricity.supply")

src.read_series("energy", key)                          # latest: newest issue, newest correction
src.read_series("energy", key, as_of=cutoff)            # only what was knowable at `cutoff`
src.read_series("energy", key, overlapping=True)        # every forecast issue kept
src.read_series("energy", key, include_updates=True)    # full audit trail (change_time, changed_by)

Frames accept and return the timedatamodel conventions: UTC microsecond timestamps, float64 values, and the pandas index contract (valid_time, or (knowledge_time, valid_time) for versioned series) — so TimeSeries.from_pandas / to_pandas round-trips directly. Only SIMPLE and VERSIONED shapes are writable; AUDIT and CORRECTED come back from reads, matching TimeSeries.validate_for_insert.

Per-warehouse settings

Each warehouse page covers the full setup: driver install, auth walkthrough, dialect-correct parameter binding, and write-path behavior.

Any setting can be passed to the factory directly, or supplied through the environment variables described in Credentials.

Use in a model

A production Predictor reads its training view from a source and stays fully backtestable:

import rebase as rb
from rebase.sources import BitemporalSpec


class DemandForecast(rb.Predictor):
    name = "demand-forecast"

    def fit(self, train) -> None:
        ...

    def predict(self, obs) -> dict:
        ...

Point the model's training loader at rb.sources.<warehouse>(connection=...).read_bitemporal(...), and the class you backtest is the class you deploy.