Implementation:TobikoData Sqlmesh Context Plan For Dev

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

Overview

Concrete plan builder for creating isolated development environments provided by SQLMesh.

Description

The Context.plan method, when used with a development environment name, creates a virtual data environment that shares unchanged models with production while only materializing modified models. It compares the current code state with the target environment, identifies changed models through fingerprint comparison, computes affected downstream models, determines minimal backfill requirements, and generates an interactive plan that can be reviewed and applied.

This implementation enables the virtual environment pattern by creating a plan that specifies which snapshots need new versions, which tables require materialization, what data needs backfilling, and how environment pointers should be updated. The create_from parameter specifies the base environment (typically production) from which to inherit unchanged snapshots, minimizing storage and compute costs.

Usage

Use this method when creating a new development environment to work on features, when updating an existing development environment after making code changes, or when you need an isolated environment for testing without affecting production or other developers. Specify create_from='prod' to base your environment on production data.

Code Reference

Source Location

• Repository: sqlmesh
• File: sqlmesh/core/context.py (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 import Context

context = Context()

I/O Contract

Inputs

Name	Type	Required	Description
environment	str	No	Name of the development environment to create/update (e.g., 'dev', 'feature_x')
start	TimeLike	No	Start date for backfill interval
end	TimeLike	No	End date for backfill interval
execution_time	TimeLike	No	Reference date/time for execution (default: now)
create_from	str	No	Base environment to inherit from (default: 'prod')
skip_tests	bool	No	Whether to skip running unit tests (default: False)
restate_models	Iterable[str]	No	Models to restate (force recomputation) for the given interval
no_gaps	bool	No	Ensure no data gaps compared to previous snapshots
skip_backfill	bool	No	Skip the backfill step entirely (default: False)
include_unmodified	bool	No	Include unmodified models in the development environment
select_models	Collection[str]	No	Filter which models to include in the plan
auto_apply	bool	No	Automatically apply the plan after creation (default: False)
no_prompts	bool	No	Disable interactive prompts (default: False)
enable_preview	bool	No	Enable preview for forward-only models in dev environments

Outputs

Name	Type	Description
plan	Plan	Plan object containing snapshot changes, backfill requirements, and environment updates

Usage Examples

Basic Development Environment Creation

from sqlmesh import Context

# Initialize context
context = Context(paths="path/to/project")

# Create a development environment based on production
plan = context.plan(
    environment="dev",
    create_from="prod",
    start="2024-01-01",
    end="2024-01-31"
)

# Review the plan interactively, then apply
context.apply(plan)

Feature Branch Development

# Create an isolated environment for a specific feature
plan = context.plan(
    environment="feature_revenue_model",
    create_from="prod",
    start="2024-01-01",
    end="2024-01-31",
    select_models=["*revenue*"],  # Only include revenue-related models
    auto_apply=True  # Automatically apply without prompts
)

print(f"Created environment with {len(plan.new_snapshots)} new snapshots")

Incremental Development Workflow

# Day 1: Create initial dev environment
context.plan(
    environment="my_dev",
    create_from="prod",
    start="2024-01-01",
    end="2024-01-07",
    auto_apply=True
)

# Day 2: Make code changes, update the same environment
# SQLMesh detects changes and updates only affected models
context.plan(
    environment="my_dev",
    start="2024-01-01",
    end="2024-01-14",  # Extended date range
    auto_apply=True
)

# The environment now has latest code with extended data

Skip Backfill for Fast Iteration

# Create environment structure without materializing data
# Useful for validating model definitions and dependencies
plan = context.plan(
    environment="test_structure",
    create_from="prod",
    skip_backfill=True,  # Don't materialize any data
    auto_apply=True
)

# Environment exists but tables are empty
# Use context.render() to validate SQL without execution

Selective Model Testing

# Test only specific models and their dependencies
plan = context.plan(
    environment="test_model_x",
    create_from="prod",
    start="2024-01-01",
    end="2024-01-01",  # Single day for fast testing
    select_models=["my_schema.specific_model", "upstream.*"],
    include_unmodified=False,  # Don't include unchanged models
    auto_apply=True
)

# Validate the results
results = context.evaluate(
    "my_schema.specific_model",
    start="2024-01-01",
    end="2024-01-01"
)
print(results)

Related Pages

Implements Principle

• Principle:TobikoData_Sqlmesh_Development_Environment_Creation