Implementation:TobikoData Sqlmesh Context Plan
| Knowledge Sources | |
|---|---|
| Domains | Data_Engineering, Deployment |
| Last Updated | 2026-02-07 00:00 GMT |
Overview
Concrete plan creation method for generating deployment plans by detecting changes and determining backfill requirements, provided by the Context class.
Description
The Context.plan method is SQLMesh's implementation of the "plan" phase in plan/apply workflows. It compares the current project state against a target environment, identifies modified models through fingerprint comparison, categorizes changes as breaking or non-breaking, determines backfill time ranges, runs unit tests, and presents an interactive summary for user review.
This method creates a PlanBuilder internally, builds the plan, and displays it via the console with change diffs and backfill estimates. Users can interactively adjust backfill ranges, categorize uncategorized changes, and approve or reject the plan. The resulting Plan object contains all information needed for the apply phase, including snapshots to promote, intervals to backfill, and deployment metadata.
The method supports extensive configuration: selective model deployment, forward-only changes for production, skipping tests or backfills, handling data gaps, restating specific models, and auto-applying plans without prompts for CI/CD workflows.
Usage
Call this method whenever deploying changes to any environment. Use it for development environments to preview impact before applying, for staging deployments with automated testing, and for production with careful review of breaking changes. Run interactively during development to adjust backfill ranges, or with auto_apply=True and no_prompts=True in CI/CD for fully automated deployments.
Code Reference
Source Location
- Repository: sqlmesh
- File: sqlmesh/core/context.py
- Class: Context
- Method: plan (lines 1311-1443)
Signature
def plan(
self,
environment: t.Optional[str] = None,
*,
start: t.Optional[TimeLike] = None,
end: t.Optional[TimeLike] = None,
execution_time: t.Optional[TimeLike] = None,
create_from: t.Optional[str] = None,
skip_tests: t.Optional[bool] = None,
restate_models: t.Optional[t.Iterable[str]] = None,
no_gaps: t.Optional[bool] = None,
skip_backfill: t.Optional[bool] = None,
empty_backfill: t.Optional[bool] = None,
forward_only: t.Optional[bool] = None,
allow_destructive_models: t.Optional[t.Collection[str]] = None,
allow_additive_models: t.Optional[t.Collection[str]] = None,
no_prompts: t.Optional[bool] = None,
auto_apply: t.Optional[bool] = None,
no_auto_categorization: t.Optional[bool] = None,
effective_from: t.Optional[TimeLike] = None,
include_unmodified: t.Optional[bool] = None,
select_models: t.Optional[t.Collection[str]] = None,
backfill_models: t.Optional[t.Collection[str]] = None,
categorizer_config: t.Optional[CategorizerConfig] = None,
enable_preview: t.Optional[bool] = None,
no_diff: t.Optional[bool] = None,
run: t.Optional[bool] = None,
diff_rendered: t.Optional[bool] = None,
skip_linter: t.Optional[bool] = None,
explain: t.Optional[bool] = None,
ignore_cron: t.Optional[bool] = None,
min_intervals: t.Optional[int] = None,
) -> Plan:
Import
from sqlmesh.core.context import Context
context = Context(paths="project")
plan = context.plan("prod")
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| environment | str | No | Target environment name (e.g., "prod", "dev"), defaults to config default |
| start | TimeLike | No | Backfill start date, defaults to earliest model data or change date |
| end | TimeLike | No | Backfill end date, defaults to now |
| execution_time | TimeLike | No | Reference time for execution, defaults to now |
| create_from | str | No | Environment to clone if target doesn't exist, defaults to "prod" |
| skip_tests | bool | No | Skip running unit tests, defaults to False |
| restate_models | Iterable[str] | No | Models or tags to restate (reprocess) for the plan interval |
| no_gaps | bool | No | Ensure no data gaps in backfill intervals, defaults to False |
| skip_backfill | bool | No | Skip backfill execution entirely, defaults to False |
| empty_backfill | bool | No | Record intervals as processed without executing, defaults to False |
| forward_only | bool | No | Apply changes only to future data (production safety), defaults to False |
| select_models | Collection[str] | No | Model selection filters (e.g., "tag:daily", "orders+") |
| backfill_models | Collection[str] | No | Models to backfill (subset of select_models) |
| no_prompts | bool | No | Disable interactive prompts, defaults to False |
| auto_apply | bool | No | Automatically apply plan after creation, defaults to False |
| no_diff | bool | No | Hide text diffs for changed models, defaults to False |
| diff_rendered | bool | No | Show diffs of rendered SQL vs raw, defaults to False |
Outputs
| Name | Type | Description |
|---|---|---|
| plan | Plan | Complete plan object with context diff, backfill requirements, and metadata |
Usage Examples
Basic Usage
from sqlmesh.core.context import Context
context = Context(paths="examples/sushi")
# Create plan for production environment
plan = context.plan("prod")
# Review plan output interactively, then apply
context.apply(plan)
Development Workflow
from sqlmesh.core.context import Context
context = Context(paths="project")
# Create development environment with limited backfill
plan = context.plan(
environment="dev_alice",
start="2024-01-01",
end="2024-01-07"
)
# Auto-apply dev plans for rapid iteration
context.apply(plan)
Production Deployment
from sqlmesh.core.context import Context
context = Context(paths="project")
# Production plan with forward-only changes (safe)
plan = context.plan(
environment="prod",
forward_only=True,
effective_from="2024-02-01",
no_gaps=True
)
# Manual review before apply
print(f"Plan has {len(plan.context_diff.modified_snapshots)} changes")
context.apply(plan)
CI/CD Automated Deployment
from sqlmesh.core.context import Context
context = Context(paths="/workspace/project")
# Fully automated plan and apply
plan = context.plan(
environment="staging",
skip_tests=False, # Always run tests in CI
no_prompts=True, # No interactive input
auto_apply=True # Apply immediately
)
# Plan is automatically applied if auto_apply=True
print(f"Deployment complete: {plan.environment_naming_info.name}")
Selective Deployment
from sqlmesh.core.context import Context
context = Context(paths="project")
# Deploy only specific models and dependencies
plan = context.plan(
environment="prod",
select_models=["tag:critical", "orders+"], # Critical tag + orders and downstream
backfill_models=["orders"], # Only backfill orders itself
start="-1 month"
)
context.apply(plan)
Restate Historical Data
from sqlmesh.core.context import Context
context = Context(paths="project")
# Reprocess specific models due to upstream correction
plan = context.plan(
environment="prod",
restate_models=["orders", "revenue_summary"],
start="2024-01-01",
end="2024-01-31"
)
context.apply(plan)