Implementation:TobikoData Sqlmesh Context Plan For Environment

Knowledge Sources	SQLMesh SQLMesh Docs
Domains	Data_Engineering, Environment_Management
Last Updated	2026-02-07 00:00 GMT

Overview

Concrete tool for creating and planning virtual data environments provided by SQLMesh.

Description

The Context.plan() method creates an execution plan for establishing or updating a virtual data environment. It compares the current project state (model definitions, configurations) against the target environment, identifies what has changed, determines which models need backfilling, and produces an actionable Plan object. When creating a new environment, it can clone from an existing environment (specified via create_from) or default to cloning from production, enabling lightweight environment branching with copy-on-write semantics.

The method orchestrates the entire planning workflow: discovering model changes through fingerprinting, categorizing changes as breaking or non-breaking, determining data intervals that need processing, running tests, validating configurations, and presenting an interactive or automated approval flow. The resulting Plan object encapsulates all operations needed to materialize the target environment state.

Usage

Use Context.plan() when you need to:

• Create a new development environment branched from production or another environment
• Update an existing environment with modified models
• Prepare a staging environment for testing before production promotion
• Generate a plan for code review showing exactly what will change
• Validate that model changes can be safely deployed
• Determine backfill requirements for incremental models
• Create PR-specific environments in CI/CD pipelines

Code Reference

Source Location

• Repository: sqlmesh
• File: sqlmesh/core/context.py:L1311-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="path/to/project")

I/O Contract

Inputs

Name	Type	Required	Description
environment	str	No	Target environment name. Defaults to configured default_target_environment
start	TimeLike	No	Start date for backfill if needed
end	TimeLike	No	End date for backfill if needed
execution_time	TimeLike	No	Reference time for execution. Defaults to current time
create_from	str	No	Source environment to clone from when creating new environment. Defaults to "prod"
skip_tests	bool	No	Skip running unit tests during plan creation
restate_models	Iterable[str]	No	Models/tags that need restating (reprocessing) for the plan interval
no_gaps	bool	No	Ensure no data gaps exist for models already in target environment
skip_backfill	bool	No	Skip the backfill step entirely
empty_backfill	bool	No	Skip backfill but record intervals as processed
forward_only	bool	No	Plan contains only forward-only changes (no breaking changes)
allow_destructive_models	Collection[str]	No	Models allowed to have destructive forward-only changes
allow_additive_models	Collection[str]	No	Models allowed to have additive forward-only changes
no_prompts	bool	No	Disable interactive prompts for backfill configuration
auto_apply	bool	No	Automatically apply plan after creation
no_auto_categorization	bool	No	Disable automatic change categorization (breaking/non-breaking)
effective_from	TimeLike	No	Effective date for applying forward-only changes in production
include_unmodified	bool	No	Include unmodified models in development environment
select_models	Collection[str]	No	Model selection strings to filter included models
backfill_models	Collection[str]	No	Model selection strings to filter models for backfilling
categorizer_config	CategorizerConfig	No	Custom configuration for change categorizer
enable_preview	bool	No	Enable preview mode for forward-only models in dev environments
no_diff	bool	No	Hide text differences for changed models
run	bool	No	Run latest intervals as part of plan application
diff_rendered	bool	No	Compare rendered SQL instead of raw model definitions
skip_linter	bool	No	Skip running linter during plan creation
explain	bool	No	Show plan explanation instead of applying
ignore_cron	bool	No	Ignore cron schedules when determining intervals
min_intervals	int	No	Minimum number of intervals to ensure coverage per model

Outputs

Name	Type	Description
return	Plan	Populated Plan object containing all changes, backfill requirements, and metadata for environment creation/update

Usage Examples

Basic Usage

from sqlmesh.core.context import Context

# Initialize context
context = Context(paths="my_project")

# Create a new development environment from production
plan = context.plan(
    environment="dev",
    create_from="prod"
)

# The plan object contains:
# - Snapshot changes (added/modified/removed models)
# - Backfill requirements
# - Test results
# - Change categorization

Creating PR Environment

from sqlmesh.core.context import Context

context = Context()

# Create isolated PR environment for testing
plan = context.plan(
    environment="pr_123",
    create_from="prod",
    skip_backfill=False,
    no_prompts=True,
    auto_apply=True
)

# Environment is created and ready for validation

Selective Model Planning

from sqlmesh.core.context import Context

context = Context()

# Plan only specific models and their downstream dependencies
plan = context.plan(
    environment="staging",
    select_models=["model_a", "tag:critical"],
    include_unmodified=False
)

# Only selected models and dependencies are included

Forward-Only Production Update

from sqlmesh.core.context import Context

context = Context()

# Plan production update with forward-only changes
plan = context.plan(
    environment="prod",
    forward_only=True,
    effective_from="2026-02-08",
    no_prompts=True
)

# Changes take effect from specified date without breaking queries

Data Gap Validation

from sqlmesh.core.context import Context

context = Context()

# Ensure no data gaps when updating production
plan = context.plan(
    environment="prod",
    no_gaps=True,
    start="2026-01-01",
    end="2026-02-07"
)

# Will fail if any gaps detected in specified interval

Testing Without Backfill

from sqlmesh.core.context import Context

context = Context()

# Create plan to see changes without running expensive backfills
plan = context.plan(
    environment="dev",
    skip_backfill=True,
    no_diff=False
)

# Review changes without data processing

Related Pages

Implements Principle

• Principle:TobikoData_Sqlmesh_Environment_Creation