Test with dbt and Soda

Two layers. Bronze you can prove from Python with no warehouse. Warehouse quality is dbt tests plus a Soda scan after every sync.

Bronze smoke

No sink required. Sync, then assert on the Arrow table.

Python

import skippr

s = skippr.Session(skippr.Config.discover().get_pipeline("bikehire"))
s.sync(once=True)
table = s.df()
assert table.num_rows > 0
assert "rider_id" in table.column_names

CLI

skipprd sync --pipeline bikehire --once --log
skipprd df --pipeline bikehire

Fast CI. The WAL is the dataset.

Warehouse path

Need a sink: Snowflake or Postgres. Models: dbt.

Python

s.sync(once=True)

CLI

skipprd sync --pipeline bikehire --once --log

Then warehouse tools:

dbt run
dbt test
soda scan -d warehouse -c configuration.yml checks.yml

dbt tests

On the silver grain. In the model repo, next to the SQL.

# models/silver/bikehire_silver.yml
version: 2

models:
  - name: bikehire_silver
    columns:
      - name: rider_id
        tests:
          - not_null
      - name: bike_id
        tests:
          - not_null

dbt test

Soda

Warehouse-level checks you can run in CI after every sync. Not dbt. Same destination skipprd just wrote.

# configuration.yml
data_source warehouse:
  type: snowflake
  connection:
    account: ${SNOWFLAKE_ACCOUNT}
    username: ${SNOWFLAKE_USER}
    database: RAW_DATA
    schema: PUBLIC
    warehouse: COMPUTE_WH

Postgres is type: postgres with host, database analytics, schema public.

# checks.yml
checks for bikehire:
  - row_count > 0
  - missing_count(rider_id) = 0

soda scan -d warehouse -c configuration.yml checks.yml

dbt tests live with the models. Soda scans the warehouse after skipprd has written. That is the same pattern skipprd uses in its own e2e: load, then Soda.