Implementation:TobikoData Sqlmesh Context Init
| Knowledge Sources | |
|---|---|
| Domains | Data_Engineering, Deployment |
| Last Updated | 2026-02-07 00:00 GMT |
Overview
Concrete initialization method for loading and validating SQLMesh projects, provided by the GenericContext class.
Description
The GenericContext.__init__ method is the entry point for creating a SQLMesh context, which encapsulates a complete SQLMesh project environment. When invoked, it discovers project files, loads configurations, parses model definitions, constructs dependency graphs, initializes database connections, and prepares the state synchronization mechanism. This method is called automatically when instantiating a Context object and completes all necessary validation before returning control to the user.
The initialization process handles multiple project paths, merges configurations with appropriate precedence, supports different loader types (standard, dbt), configures gateway connections to data warehouses, and optionally defers loading via the load parameter for use cases requiring programmatic configuration before project discovery.
Usage
Import and instantiate Context at the beginning of any SQLMesh workflow. Use paths parameter to specify project directories, config parameter to provide configuration overrides, gateway parameter to select a specific connection, and load=False when you need to configure the context programmatically before loading models. This is the first call in CLI commands, API scripts, notebook environments, and automated CI/CD pipelines.
Code Reference
Source Location
- Repository: sqlmesh
- File: sqlmesh/core/context.py
- Class: GenericContext
- Method: __init__ (lines 376-474)
Signature
def __init__(
self,
notification_targets: t.Optional[t.List[NotificationTarget]] = None,
state_sync: t.Optional[StateSync] = None,
paths: t.Union[str | Path, t.Iterable[str | Path]] = "",
config: t.Optional[t.Union[C, str, t.Dict[Path, C]]] = None,
gateway: t.Optional[str] = None,
concurrent_tasks: t.Optional[int] = None,
loader: t.Optional[t.Type[Loader]] = None,
load: bool = True,
users: t.Optional[t.List[User]] = None,
config_loader_kwargs: t.Optional[t.Dict[str, t.Any]] = None,
selector: t.Optional[t.Type[Selector]] = None,
):
Import
from sqlmesh.core.context import Context
# Standard usage
context = Context(paths="path/to/project")
# With configuration override
context = Context(paths="project", config="config.yaml")
# Multiple paths
context = Context(paths=["project1", "project2"])
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| notification_targets | t.List[NotificationTarget] | No | Custom notification targets for plan/apply events |
| state_sync | StateSync | No | Pre-configured state sync instance (advanced usage) |
| paths | str, Path, or Iterable | No | Project directory path(s) to load, defaults to current directory |
| config | Config, str, or Dict | No | Configuration object, file path, or per-path config mapping |
| gateway | str | No | Gateway name to use for database connections |
| concurrent_tasks | int | No | Number of concurrent tasks for parallel operations |
| loader | Type[Loader] | No | Custom loader class (e.g., dbt loader) |
| load | bool | No | Whether to immediately load project, defaults to True |
| users | t.List[User] | No | User definitions for access control |
| config_loader_kwargs | Dict[str, Any] | No | Additional arguments for config loading |
| selector | Type[Selector] | No | Custom selector class for model selection |
Outputs
| Name | Type | Description |
|---|---|---|
| self | GenericContext | Initialized context instance with loaded project state |
Usage Examples
Basic Usage
from sqlmesh.core.context import Context
# Load project from current directory
context = Context()
# Load project from specific path
context = Context(paths="examples/sushi")
# Use specific gateway connection
context = Context(paths="project", gateway="snowflake")
Advanced Usage
from sqlmesh.core.context import Context
from sqlmesh.core.config import Config
# Load with custom configuration
config = Config(
model_defaults={"dialect": "duckdb"},
default_connection={"type": "duckdb", "database": ":memory:"}
)
context = Context(paths="project", config=config)
# Defer loading for programmatic setup
context = Context(paths="project", load=False)
# Configure context...
context.load()
# Multiple project paths
context = Context(paths=["shared_models", "team_models"])
CI/CD Usage
from sqlmesh.core.context import Context
# Initialize for automated testing
context = Context(
paths="/workspace/sqlmesh_project",
gateway="ci_warehouse",
concurrent_tasks=4
)
# Run tests before deployment
result = context.test()
if result.failures:
raise Exception("Tests failed")
# Create and apply plan
plan = context.plan("prod", auto_apply=True)
context.apply(plan)