dbt Renderer

Compiles dbt models via subprocess (dbt compile), then parses the compiled SQL with SqlParser.

DbtRenderer

DbtRenderer(sql_parser=None)

Compiles dbt models via dbt compile and parses the resulting SQL.

Shells out to dbt (via uv run or a custom command) in the project's own virtualenv, then reads compiled SQL from target/compiled/ and feeds each model through SqlParser. dbt is not a Python dependency of the indexer -- it uses whatever version the project has installed.

Initialise the renderer.

Parameters:

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

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

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

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

render_project

render_project(
    project_path,
    profiles_dir=None,
    env_file=None,
    target=None,
    dbt_command="uv run dbt",
    venv_dir=None,
    dialect=None,
    schema_catalog=None,
)

Compile all dbt models and parse the resulting SQL.

Parameters:

Name	Type	Description	Default
project_path	str | Path	Path to dbt project dir (containing dbt_project.yml)	required
profiles_dir	str | Path | None	Path to directory containing profiles.yml (defaults to project_path)	None
env_file	str | Path | None	Optional .env file to source before running dbt compile	None
target	str | None	dbt target name (default: whatever profiles.yml specifies)	None
dbt_command	str	Command to invoke dbt (default: "uv run dbt")	'uv run dbt'
venv_dir	str | Path | None	Directory to run uv run from (where .venv lives). Defaults to project_path, but auto-detects parent if parent has .venv and project_path doesn't.	None
dialect	str | None	SQL dialect for parsing (e.g. "starrocks", "mysql", "postgres"). Needed for dialect-specific syntax like backtick quoting.	None

Returns:

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

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

def render_project(
    self,
    project_path: str | Path,
    profiles_dir: str | Path | None = None,
    env_file: str | Path | None = None,
    target: str | None = None,
    dbt_command: str = "uv run dbt",
    venv_dir: str | Path | None = None,
    dialect: str | None = None,
    schema_catalog: dict | None = None,
) -> dict[str, ParseResult]:
    """Compile all dbt models and parse the resulting SQL.

    Args:
        project_path: Path to dbt project dir (containing dbt_project.yml)
        profiles_dir: Path to directory containing profiles.yml (defaults to project_path)
        env_file: Optional .env file to source before running dbt compile
        target: dbt target name (default: whatever profiles.yml specifies)
        dbt_command: Command to invoke dbt (default: "uv run dbt")
        venv_dir: Directory to run `uv run` from (where .venv lives).
                  Defaults to project_path, but auto-detects parent if
                  parent has .venv and project_path doesn't.
        dialect: SQL dialect for parsing (e.g. "starrocks", "mysql", "postgres").
                 Needed for dialect-specific syntax like backtick quoting.

    Returns:
        Dict mapping model relative path -> ParseResult
    """
    return self._compile_and_parse(
        project_path=project_path,
        profiles_dir=profiles_dir,
        env_file=env_file,
        target=target,
        dbt_command=dbt_command,
        venv_dir=venv_dir,
        dialect=dialect,
        schema_catalog=schema_catalog,
    )

render_models

render_models(
    project_path,
    model_names,
    profiles_dir=None,
    env_file=None,
    target=None,
    dbt_command="uv run dbt",
    venv_dir=None,
    dialect=None,
    schema_catalog=None,
)

Compile specific dbt models using --select and parse the resulting SQL.

Parameters:

Name	Type	Description	Default
project_path	str | Path	Path to dbt project dir (containing dbt_project.yml)	required
model_names	list[str]	List of model names to compile (passed to --select)	required
profiles_dir	str | Path | None	Path to directory containing profiles.yml (defaults to project_path)	None
env_file	str | Path | None	Optional .env file to source before running dbt compile	None
target	str | None	dbt target name (default: whatever profiles.yml specifies)	None
dbt_command	str	Command to invoke dbt (default: "uv run dbt")	'uv run dbt'
venv_dir	str | Path | None	Directory to run uv run from (where .venv lives).	None
dialect	str | None	SQL dialect for parsing.	None
schema_catalog	dict | None	Optional schema catalog for column resolution.	None

Returns:

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

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

def render_models(
    self,
    project_path: str | Path,
    model_names: list[str],
    profiles_dir: str | Path | None = None,
    env_file: str | Path | None = None,
    target: str | None = None,
    dbt_command: str = "uv run dbt",
    venv_dir: str | Path | None = None,
    dialect: str | None = None,
    schema_catalog: dict | None = None,
) -> dict[str, ParseResult]:
    """Compile specific dbt models using ``--select`` and parse the resulting SQL.

    Args:
        project_path: Path to dbt project dir (containing dbt_project.yml)
        model_names: List of model names to compile (passed to ``--select``)
        profiles_dir: Path to directory containing profiles.yml (defaults to project_path)
        env_file: Optional .env file to source before running dbt compile
        target: dbt target name (default: whatever profiles.yml specifies)
        dbt_command: Command to invoke dbt (default: "uv run dbt")
        venv_dir: Directory to run `uv run` from (where .venv lives).
        dialect: SQL dialect for parsing.
        schema_catalog: Optional schema catalog for column resolution.

    Returns:
        Dict mapping model relative path -> ParseResult
    """
    return self._compile_and_parse(
        project_path=project_path,
        profiles_dir=profiles_dir,
        env_file=env_file,
        target=target,
        dbt_command=dbt_command,
        venv_dir=venv_dir,
        dialect=dialect,
        schema_catalog=schema_catalog,
        select=model_names,
    )

extract_schema_yml

extract_schema_yml(project_path)

Extract column definitions from dbt schema.yml files.

Scans all *.yml and *.yaml files under the project's models/ directory for model and source entries with columns: lists. Returns a mapping of model name to ColumnDefResult entries with source='schema_yml'. Sources are keyed as "{source_family}.{table_name}" for backwards compat with the pre-existing catalog flatten.

Parameters:

Name	Type	Description	Default
project_path	str | Path	Path to dbt project dir (containing models/).	required

Returns:

Type	Description
dict[str, list[ColumnDefResult]]	Dict mapping model name -> list of ColumnDefResult.

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

def extract_schema_yml(
    self, project_path: str | Path
) -> dict[str, list[ColumnDefResult]]:
    """Extract column definitions from dbt schema.yml files.

    Scans all ``*.yml`` and ``*.yaml`` files under the project's ``models/``
    directory for model and source entries with ``columns:`` lists. Returns
    a mapping of model name to ``ColumnDefResult`` entries with
    ``source='schema_yml'``. Sources are keyed as
    ``"{source_family}.{table_name}"`` for backwards compat with the
    pre-existing catalog flatten.

    Args:
        project_path: Path to dbt project dir (containing ``models/``).

    Returns:
        Dict mapping model name -> list of ``ColumnDefResult``.
    """
    models, sources = self._extract_schema_yml_by_kind(project_path)
    result: dict[str, list[ColumnDefResult]] = dict(models)
    for src in sources:
        key = (
            f"{src.family}.{src.identifier}" if src.family else src.identifier
        )
        existing = result.get(key)
        if existing is None:
            result[key] = list(src.columns)
        else:
            existing.extend(src.columns)
    return result