Implementation:TobikoData Sqlmesh Load Sql Based Model

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

Overview

Concrete parser and loader for SQL-based model definitions provided by SQLMesh.

Description

The load_sql_based_model function parses SQLMesh model files written in SQL with embedded MODEL annotations. It processes a list of parsed expressions (typically containing a MODEL configuration block followed by SQL statements), validates the configuration, normalizes the model name, resolves the materialization kind, infers dependencies from the query, and constructs a Model object with all metadata properly initialized.

This function handles the complexity of merging default configurations with explicit model settings, expanding macros, validating syntax, and ensuring all required metadata (particularly the model name) is present. It supports various model kinds including FULL, INCREMENTAL_BY_TIME_RANGE, VIEW, and SCD_TYPE_2, each with different materialization and processing semantics.

Usage

Use this function when programmatically loading SQLMesh model definitions from parsed SQL files, when building custom loaders or integrations, or when you need fine-grained control over model loading with specific defaults and macro registries. The Context class typically calls this function internally when loading models from a project directory.

Code Reference

Source Location

• Repository: sqlmesh
• File: sqlmesh/core/model/definition.py (Lines 2162-2209)

Signature

def load_sql_based_model(
    expressions: t.List[exp.Expression],
    *,
    defaults: t.Optional[t.Dict[str, t.Any]] = None,
    path: t.Optional[Path] = None,
    module_path: Path = Path(),
    time_column_format: str = c.DEFAULT_TIME_COLUMN_FORMAT,
    macros: t.Optional[MacroRegistry] = None,
    jinja_macros: t.Optional[JinjaMacroRegistry] = None,
    audits: t.Optional[t.Dict[str, ModelAudit]] = None,
    python_env: t.Optional[t.Dict[str, Executable]] = None,
    dialect: t.Optional[str] = None,
    physical_schema_mapping: t.Optional[t.Dict[re.Pattern, str]] = None,
    default_catalog: t.Optional[str] = None,
    variables: t.Optional[t.Dict[str, t.Any]] = None,
    infer_names: t.Optional[bool] = False,
    blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None,
    **kwargs: t.Any,
) -> Model

Import

from sqlmesh.core.model import load_sql_based_model

I/O Contract

Inputs

Name	Type	Required	Description
expressions	List[exp.Expression]	Yes	Parsed SQLGlot expressions containing MODEL block and query statements
defaults	Dict[str, Any]	No	Default configuration values to merge with model-specific settings
path	Path	No	File path to the model definition for error reporting
module_path	Path	No	Python module path for macro serialization (default: Path())
time_column_format	str	No	Default time column format (default: c.DEFAULT_TIME_COLUMN_FORMAT)
macros	MacroRegistry	No	Custom macro registry (default: global registry)
jinja_macros	JinjaMacroRegistry	No	Jinja macro registry for template expansion
audits	Dict[str, ModelAudit]	No	Available audit definitions to attach to models
python_env	Dict[str, Executable]	No	Python environment for macro execution
dialect	str	No	SQL dialect for parsing and rendering (e.g., 'snowflake', 'bigquery')
physical_schema_mapping	Dict[re.Pattern, str]	No	Regex patterns mapping logical to physical schemas
default_catalog	str	No	Default catalog name for three-part model naming
variables	Dict[str, Any]	No	User-defined variables accessible in model queries
infer_names	bool	No	Whether to infer model names from context (default: False)
blueprint_variables	Dict[str, Any]	No	Variables specific to blueprint model templates

Outputs

Name	Type	Description
model	Model	Fully initialized Model object with parsed query, metadata, and dependencies

Usage Examples

Basic Usage

from sqlglot import parse
from sqlmesh.core.model import load_sql_based_model

# Parse a SQLMesh model SQL file
sql_content = """
MODEL (
  name my_schema.my_model,
  kind FULL,
  cron '@daily',
  audits [assert_positive_values]
);

SELECT
  id,
  revenue,
  created_at
FROM raw.transactions
WHERE revenue > 0;
"""

expressions = parse(sql_content, read='sqlmesh')

# Load the model with defaults
model = load_sql_based_model(
    expressions=expressions,
    defaults={'dialect': 'snowflake'},
    dialect='snowflake',
    default_catalog='my_catalog'
)

print(f"Model name: {model.name}")
print(f"Model kind: {model.kind}")
print(f"Schedule: {model.cron}")

Advanced Usage with Custom Macros

from pathlib import Path
from sqlmesh.core.macros import MacroRegistry
from sqlmesh.core.model import load_sql_based_model

# Create custom macro registry
macro_registry = MacroRegistry()

@macro_registry.register("my_custom_macro")
def custom_transformation(evaluator, *args):
    # Custom macro logic
    return "transformed_value"

# Load model with custom configuration
model = load_sql_based_model(
    expressions=parsed_expressions,
    path=Path("models/my_model.sql"),
    macros=macro_registry,
    defaults={
        'dialect': 'bigquery',
        'owner': 'data-team'
    },
    time_column_format='%Y-%m-%d',
    default_catalog='production'
)

# Access model properties
print(f"Dependencies: {model.depends_on}")
print(f"Columns: {model.columns_to_types}")

Related Pages

Implements Principle

• Principle:TobikoData_Sqlmesh_Model_Definition