SQLMesh is the next-generation SQL transformation framework with virtual environments for safe deploys. pip install sqlmesh. Models in models/ directory: MODEL (name db.table, kind FULL) for full refresh, MODEL (name db.orders_daily, kind INCREMENTAL_BY_UNIQUE_KEY (unique_key order_id)) for upserts, MODEL (name db.events, kind INCREMENTAL_BY_TIME_RANGE (time_column event_date)) for time-partitioned incremental. sqlmesh plan computes a change plan comparing current vs prod state — shows which models are affected, previews virtual schema changes before apply. sqlmesh plan --auto-apply deploys without confirmation. sqlmesh run executes backfills. @this_model in SQL refers to the current model’s table. @start_dt / @end_dt are the current execution interval. Python models: from sqlmesh import model, @model(name="db.enriched_users", columns={"id": "text", "score": "double"}), def execute(context, start, end, execution_time, **kwargs) -> pd.DataFrame. Audits: AUDIT (name assert_positive_amount) SELECT * FROM @this_model WHERE amount <= 0 — fails if rows returned. Metrics: METRIC (name revenue, owner "finance") SELECT SUM(amount) AS revenue FROM db.orders. Environments: sqlmesh plan dev creates a virtual dev__ schema — queries same underlying tables without copying data. sqlmesh test runs model tests. sqlmesh diff prod dev shows semantic diff between environments. Claude Code generates SQLMesh model definitions, audit tests, Python models, and plan/apply workflows.
CLAUDE.md for SQLMesh
## SQLMesh Stack
- Version: sqlmesh >= 0.100
- Config: config.yaml — connection, state_connection, default_target_environment
- Models: models/*.sql or models/*.py — MODEL() header required
- Plan: sqlmesh plan [env] — preview changes; sqlmesh plan --auto-apply to deploy
- Run: sqlmesh run [env] — execute backfills for pending intervals
- Envs: sqlmesh plan dev — creates virtual dev__ schema for safe testing
- Test: sqlmesh test -- unit tests in tests/ directory (YAML format)
- Audit: AUDIT blocks in model SQL — fail model if rows returnedSQL Models
-- models/staging/stg_orders.sql — staging model with type casting
MODEL (
name staging.stg_orders,
kind FULL,
grains [order_id],
description 'Cleaned and typed orders from raw source',
owner 'data-eng',
tags ['staging', 'finance'],
column_descriptions (
order_id 'Unique order identifier',
user_id 'FK to users table',
amount_usd 'Order amount in USD',
status 'Order lifecycle status',
created_at 'Order creation timestamp UTC'
)
);
SELECT
CAST(order_id AS TEXT) AS order_id,
CAST(user_id AS TEXT) AS user_id,
CAST(amount AS DOUBLE) AS amount_usd,
LOWER(TRIM(status)) AS status,
CAST(created_at AS TIMESTAMP) AS created_at,
CAST(updated_at AS TIMESTAMP) AS updated_at,
DATE_TRUNC('day', CAST(created_at AS TIMESTAMP)) AS created_date,
CURRENT_TIMESTAMP AS _loaded_at
FROM raw.orders
WHERE
order_id IS NOT NULL
AND amount IS NOT NULL
AND amount > 0;-- models/marts/orders_daily.sql — incremental model by unique key
MODEL (
name marts.orders_daily,
kind INCREMENTAL_BY_UNIQUE_KEY (
unique_key order_id
),
grains [order_id],
cron '@daily',
owner 'data-eng',
tags ['mart', 'finance'],
audits [assert_positive_amounts, assert_valid_status],
description 'Daily order snapshot with latest status',
);
SELECT
o.order_id,
o.user_id,
o.amount_usd,
o.status,
o.created_date,
o.created_at,
o.updated_at,
u.email AS user_email,
u.plan AS user_plan,
u.country AS user_country
FROM staging.stg_orders o
LEFT JOIN staging.stg_users u
ON o.user_id = u.user_id
WHERE o.updated_at BETWEEN @start_dt AND @end_dt;-- models/marts/revenue_hourly.sql — time-range incremental for analytics
MODEL (
name marts.revenue_hourly,
kind INCREMENTAL_BY_TIME_RANGE (
time_column event_hour,
batch_size 1, /* process 1 day at a time */
batch_concurrency 4 /* 4 days in parallel */
),
cron '@hourly',
grain event_hour,
partitioned_by event_hour,
clustered_by [user_country],
description 'Hourly revenue aggregates for dashboards',
);
SELECT
DATE_TRUNC('hour', created_at) AS event_hour,
user_country,
user_plan,
COUNT(*) AS order_count,
SUM(amount_usd) AS total_revenue,
AVG(amount_usd) AS avg_order_value,
COUNT(DISTINCT user_id) AS unique_buyers
FROM marts.orders_daily
WHERE
created_at >= @start_dt
AND created_at < @end_dt
AND status = 'completed'
GROUP BY 1, 2, 3;Audit Tests
-- audits/assert_positive_amounts.sql — audit that fails if rows returned
AUDIT (
name assert_positive_amounts,
dialect spark,
description 'Ensure all order amounts are positive'
);
SELECT *
FROM @this_model
WHERE amount_usd <= 0;-- audits/assert_valid_status.sql
AUDIT (
name assert_valid_status,
description 'Ensure status is in allowed values'
);
SELECT *
FROM @this_model
WHERE status NOT IN ('pending', 'processing', 'completed', 'refunded', 'cancelled');Python Model
# models/ml/churn_features.py — Python model for complex feature engineering
import pandas as pd
from sqlmesh import model
@model(
name="ml.churn_features",
columns={
"user_id": "text",
"days_since_last_order": "int",
"order_count_30d": "int",
"total_spend_90d": "double",
"avg_order_value": "double",
"churn_label": "boolean",
},
description="Feature table for churn prediction model",
owner="ml-team",
tags=["ml", "features"],
cron="@daily",
)
def execute(
context,
start: pd.Timestamp,
end: pd.Timestamp,
**kwargs,
) -> pd.DataFrame:
"""Build churn features from orders data."""
# Read from upstream models
orders = context.fetchdf(
"SELECT user_id, amount_usd, created_at, status FROM marts.orders_daily"
)
if orders.empty:
return pd.DataFrame(columns=list(
context.current_model.columns # type: ignore
))
orders["created_at"] = pd.to_datetime(orders["created_at"], utc=True)
now = pd.Timestamp(end, tz="UTC")
completed = orders[orders["status"] == "completed"].copy()
features = (
completed.groupby("user_id")
.agg(
last_order_date = ("created_at", "max"),
order_count_30d = ("created_at", lambda x: (x >= now - pd.Timedelta(days=30)).sum()),
total_spend_90d = ("amount_usd", lambda x: x[completed.loc[x.index, "created_at"] >= now - pd.Timedelta(days=90)].sum()),
avg_order_value = ("amount_usd", "mean"),
)
.reset_index()
)
features["days_since_last_order"] = (now - features["last_order_date"]).dt.days
features["churn_label"] = features["days_since_last_order"] > 60
return features[["user_id", "days_since_last_order", "order_count_30d",
"total_spend_90d", "avg_order_value", "churn_label"]]config.yaml
# config.yaml — SQLMesh project configuration
gateways:
local:
connection:
type: duckdb
database: data/dev.db
production:
connection:
type: bigquery
project: my-gcp-project
location: US
state_connection:
type: bigquery
project: my-gcp-project
dataset: sqlmesh_state
default_gateway: local
model_defaults:
dialect: bigquery # target SQL dialect (transpiled by SQLGlot)
cron: "@daily"
audits: []
owner: "data-engineering"
plan:
forward_only: false
auto_categorize_changes:
external: full
# CI/CD: environments map to schemas
environments:
dev: { suffix_target: schema }
staging: { suffix_target: schema }
prod: {} # no suffix — deploys to production schemasFor the dbt Core alternative when needing the largest community ecosystem (100+ packages on dbt Hub), Jinja templating for macro-heavy pipelines, dbt Cloud Semantic Layer, native support on Databricks/Snowflake managed services, and maximum operator familiarity across analytics engineering teams — dbt is the incumbent while SQLMesh is the modernized successor with virtual schemas for zero-copy environments, native Python models without dbt-core plugins, built-in audit blocks, and safer CI deployments with column-level lineage. For the Dataform alternative when working within Google Cloud and wanting a tightly integrated workflow with BigQuery, GitHub, and dbt-like SQL with SQLX syntax — Dataform is deeply integrated into the GCP Console while SQLMesh is open-source and dialect-agnostic via SQLGlot. The Claude Skills 360 bundle includes SQLMesh skill sets covering SQL models, incremental strategies, Python models, and audit tests. Start with the free tier to try analytics engineering generation.