SQLMesh Renderer

Renders SQLMesh models into clean SQL via subprocess, then parses with SqlParser.

SqlMeshRenderer

SqlMeshRenderer(sql_parser=None)

Renders sqlmesh models into ParseResult objects via subprocess.

Runs an inline Python script inside the sqlmesh project's own virtualenv to load the project, render every model to SQL, and output JSON to stdout. The rendered SQL is then parsed by SqlParser. This avoids requiring sqlmesh as a direct dependency of the indexer.

Initialise the renderer.

Parameters:

Name	Type	Description	Default
sql_parser	SqlParser | None	SqlParser instance to use for parsing rendered SQL. Creates a default instance if not provided.	None

Source code in src/sqlprism/languages/sqlmesh.py

def __init__(self, sql_parser: SqlParser | None = None):
    """Initialise the renderer.

    Args:
        sql_parser: ``SqlParser`` instance to use for parsing rendered SQL.
            Creates a default instance if not provided.
    """
    self.sql_parser = sql_parser or SqlParser()

render_project_raw

render_project_raw(
    project_path,
    env_file=None,
    variables=None,
    gateway="local",
    dialect="athena",
    sqlmesh_command="uv run python",
    venv_dir=None,
)

Render all models and return raw SQL + column schemas (no parsing).

Returns:

Type	Description
dict[str, str]	Tuple of (models_dict, column_schemas_dict) where models maps
dict[str, dict[str, str]]	model_name -> rendered_sql and column_schemas maps
tuple[dict[str, str], dict[str, dict[str, str]]]	model_name -> {col_name: data_type}.

Source code in src/sqlprism/languages/sqlmesh.py

def render_project_raw(
    self,
    project_path: str | Path,
    env_file: str | Path | None = None,
    variables: dict[str, str | int] | None = None,
    gateway: str = "local",
    dialect: str = "athena",
    sqlmesh_command: str = "uv run python",
    venv_dir: str | Path | None = None,
) -> tuple[dict[str, str], dict[str, dict[str, str]]]:
    """Render all models and return raw SQL + column schemas (no parsing).

    Returns:
        Tuple of (models_dict, column_schemas_dict) where models maps
        model_name -> rendered_sql and column_schemas maps
        model_name -> {col_name: data_type}.
    """
    project_path = Path(project_path).resolve()
    cwd = Path(venv_dir).resolve() if venv_dir else find_venv_dir(project_path)
    env = build_env(env_file)
    vars_ = variables or {}

    try:
        all_models = self._list_models(
            project_path, cwd, env, vars_, gateway, dialect, sqlmesh_command,
        )
    except Exception:
        logger.warning("Model discovery failed, falling back to single subprocess", exc_info=True)
        all_models = []

    if len(all_models) >= 20:
        models, errors, column_schemas = self._render_batches_parallel(
            project_path, cwd, env, vars_, gateway, dialect,
            sqlmesh_command, all_models,
        )
    else:
        models, errors, column_schemas = self._run_render_script(
            project_path=project_path, cwd=cwd, env=env,
            variables=vars_, gateway=gateway, dialect=dialect,
            sqlmesh_command=sqlmesh_command, model_filter=all_models if all_models else [],
        )

    for err in errors:
        logger.warning(
            "sqlmesh render error for model %s: %s",
            err.get("model", "<unknown>"),
            err.get("error", "<no message>"),
        )

    return models, column_schemas

render_project

render_project(
    project_path,
    env_file=None,
    variables=None,
    gateway="local",
    dialect="athena",
    sqlmesh_command="uv run python",
    venv_dir=None,
    schema_catalog=None,
)

Render all models in a sqlmesh project.

Parameters:

Name	Type	Description	Default
project_path	str | Path	Path to the sqlmesh project directory (containing config.yaml)	required
env_file	str | Path | None	Path to .env file to source before loading context	None
variables	dict[str, str | int] | None	Extra sqlmesh variables (e.g. {"GRACE_PERIOD": 7})	None
gateway	str	Gateway name to use (default "local" — uses duckdb, no remote deps)	'local'
dialect	str	SQL dialect for rendering output	'athena'
sqlmesh_command	str	Command to run python in the sqlmesh venv (default: "uv run python")	'uv run python'
venv_dir	str | Path | None	Directory to run from (where .venv lives). Auto-detects if not set.	None

Returns:

Type	Description
dict[str, ParseResult]	Dict mapping model name -> ParseResult

Source code in src/sqlprism/languages/sqlmesh.py

def render_project(
    self,
    project_path: str | Path,
    env_file: str | Path | None = None,
    variables: dict[str, str | int] | None = None,
    gateway: str = "local",
    dialect: str = "athena",
    sqlmesh_command: str = "uv run python",
    venv_dir: str | Path | None = None,
    schema_catalog: dict | None = None,
) -> dict[str, ParseResult]:
    """Render all models in a sqlmesh project.

    Args:
        project_path: Path to the sqlmesh project directory (containing config.yaml)
        env_file: Path to .env file to source before loading context
        variables: Extra sqlmesh variables (e.g. {"GRACE_PERIOD": 7})
        gateway: Gateway name to use (default "local" — uses duckdb, no remote deps)
        dialect: SQL dialect for rendering output
        sqlmesh_command: Command to run python in the sqlmesh venv (default: "uv run python")
        venv_dir: Directory to run from (where .venv lives). Auto-detects if not set.

    Returns:
        Dict mapping model name -> ParseResult
    """
    return self._render_and_parse(
        project_path=project_path,
        env_file=env_file,
        variables=variables,
        gateway=gateway,
        dialect=dialect,
        sqlmesh_command=sqlmesh_command,
        venv_dir=venv_dir,
        schema_catalog=schema_catalog,
    )

render_models

render_models(
    project_path,
    model_names,
    env_file=None,
    variables=None,
    gateway="local",
    dialect="athena",
    sqlmesh_command="uv run python",
    venv_dir=None,
    schema_catalog=None,
)

Render specific models in a sqlmesh project.

Parameters:

Name	Type	Description	Default
project_path	str | Path	Path to the sqlmesh project directory	required
model_names	list[str]	List of model names to render (passed as filter to render script)	required
env_file	str | Path | None	Path to .env file to source before loading context	None
variables	dict[str, str | int] | None	Extra sqlmesh variables	None
gateway	str	Gateway name to use (default "local")	'local'
dialect	str	SQL dialect for rendering output	'athena'
sqlmesh_command	str	Command to run python in the sqlmesh venv	'uv run python'
venv_dir	str | Path | None	Directory to run from (where .venv lives)	None
schema_catalog	dict | None	Optional schema catalog for column resolution	None

Returns:

Type	Description
dict[str, ParseResult]	Dict mapping model name -> ParseResult

Source code in src/sqlprism/languages/sqlmesh.py

def render_models(
    self,
    project_path: str | Path,
    model_names: list[str],
    env_file: str | Path | None = None,
    variables: dict[str, str | int] | None = None,
    gateway: str = "local",
    dialect: str = "athena",
    sqlmesh_command: str = "uv run python",
    venv_dir: str | Path | None = None,
    schema_catalog: dict | None = None,
) -> dict[str, ParseResult]:
    """Render specific models in a sqlmesh project.

    Args:
        project_path: Path to the sqlmesh project directory
        model_names: List of model names to render (passed as filter to render script)
        env_file: Path to .env file to source before loading context
        variables: Extra sqlmesh variables
        gateway: Gateway name to use (default "local")
        dialect: SQL dialect for rendering output
        sqlmesh_command: Command to run python in the sqlmesh venv
        venv_dir: Directory to run from (where .venv lives)
        schema_catalog: Optional schema catalog for column resolution

    Returns:
        Dict mapping model name -> ParseResult
    """
    return self._render_and_parse(
        project_path=project_path,
        env_file=env_file,
        variables=variables,
        gateway=gateway,
        dialect=dialect,
        sqlmesh_command=sqlmesh_command,
        venv_dir=venv_dir,
        schema_catalog=schema_catalog,
        model_filter=model_names,
    )