Implementation:TobikoData Sqlmesh DbtContext Add Macros

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

Overview

Concrete tool for registering dbt Jinja macros in SQLMesh's macro environment provided by SQLMesh.

Description

The DbtContext class provides the environment for executing dbt models within SQLMesh, maintaining Jinja macro registries, variable scopes, and model references. The add_macros method registers dbt macros (both built-in and custom) into the JinjaMacroRegistry, organizing them by package namespace for proper resolution during model rendering. This method enables SQLMesh to execute dbt's Jinja templating while preserving package-scoped macro visibility and override behavior.

The context handles the complexities of dbt's macro system including package-qualified references, macro dispatch (selecting implementations based on adapter type), and the interaction between macros and variables. It maintains a Jinja environment that gets rebuilt when macros, models, or variables change, ensuring consistent rendering behavior.

Usage

Use DbtContext and its add_macros method when building the Jinja environment for dbt model rendering. This is typically done during the project loading phase when macros are discovered and need to be registered for use in model queries. The context ensures macros are available with correct package scoping for both ref() style references and package-qualified macro calls.

Code Reference

Source Location

• Repository: sqlmesh
• File: sqlmesh/dbt/context.py

Signature

@dataclass
class DbtContext:
    def __init__(
        project_root: Path = Path(),
        profiles_dir: t.Optional[Path] = None,
        target_name: t.Optional[str] = None,
        profile_name: t.Optional[str] = None,
        project_schema: t.Optional[str] = None,
        jinja_macros: JinjaMacroRegistry = ...,
        sqlmesh_config: SQLMeshConfig = ...,
        ...
    ):
        ...

    def add_macros(self, macros: t.Dict[str, MacroInfo], package: str) -> None:
        ...

Import

from pathlib import Path
from sqlmesh.dbt.context import DbtContext
from sqlmesh.core.macros import MacroInfo

I/O Contract

Inputs

Name	Type	Required	Description
project_root	Path	No	Path to dbt project root directory; defaults to current directory
profiles_dir	Path	No	Path to directory containing profiles.yml; defaults to ~/.dbt
target_name	str	No	Target name from profile to use
profile_name	str	No	Profile name from profiles.yml
jinja_macros	JinjaMacroRegistry	No	Pre-configured macro registry; defaults to new registry with dbt built-ins
macros (add_macros)	Dict[str, MacroInfo]	Yes	Dictionary mapping macro names to MacroInfo objects
package (add_macros)	str	Yes	Package name for namespace scoping

Outputs

Name	Type	Description
None	None	add_macros modifies the context in-place, invalidating cached Jinja environment

Usage Examples

Basic Macro Registration

from pathlib import Path
from sqlmesh.dbt.context import DbtContext
from sqlmesh.core.macros import MacroInfo

# Create DbtContext
context = DbtContext(
    project_root=Path("/dbt/project"),
    profile_name="my_profile",
    target_name="dev"
)

# Define macro information
macros = {
    "cents_to_dollars": MacroInfo(
        name="cents_to_dollars",
        definition="{% macro cents_to_dollars(column_name) %}{{ column_name }} / 100{% endmacro %}",
        depends_on=[]
    ),
    "generate_alias": MacroInfo(
        name="generate_alias",
        definition="{% macro generate_alias(custom_alias) %}{{ custom_alias or '' }}{% endmacro %}",
        depends_on=[]
    )
}

# Register macros for project package
context.add_macros(macros, package="my_project")

# Macros are now available for rendering
rendered = context.render("{{ cents_to_dollars('price_cents') }}")
print(rendered)  # Output: price_cents / 100

Multi-Package Macro Registration

from pathlib import Path
from sqlmesh.dbt.context import DbtContext
from sqlmesh.core.macros import MacroInfo

context = DbtContext(project_root=Path("/dbt/project"))
context.project_name = "my_project"

# Register root project macros
project_macros = {
    "custom_schema": MacroInfo(
        name="custom_schema",
        definition="{% macro custom_schema(schema) %}{{ schema }}_custom{% endmacro %}",
        depends_on=[]
    )
}
context.add_macros(project_macros, package="my_project")

# Register package dependency macros
dbt_utils_macros = {
    "surrogate_key": MacroInfo(
        name="surrogate_key",
        definition="{% macro surrogate_key(fields) %}{{ fields | join('||') }}{% endmacro %}",
        depends_on=[]
    )
}
context.add_macros(dbt_utils_macros, package="dbt_utils")

# Use package-qualified reference
rendered = context.render("{{ dbt_utils.surrogate_key(['id', 'created_at']) }}")
print(rendered)

# Or use unqualified if it's in current package
rendered = context.render("{{ custom_schema('staging') }}")
print(rendered)  # Output: staging_custom

Macro with Dependencies

from pathlib import Path
from sqlmesh.dbt.context import DbtContext
from sqlmesh.core.macros import MacroInfo

context = DbtContext(project_root=Path("/dbt/project"))
context.project_name = "my_project"

# Register helper macro first
helper_macros = {
    "get_column_name": MacroInfo(
        name="get_column_name",
        definition="{% macro get_column_name() %}revenue{% endmacro %}",
        depends_on=[]
    )
}
context.add_macros(helper_macros, package="my_project")

# Register macro that depends on helper
dependent_macros = {
    "calculate_total": MacroInfo(
        name="calculate_total",
        definition="{% macro calculate_total() %}SUM({{ get_column_name() }}){% endmacro %}",
        depends_on=["get_column_name"]
    )
}
context.add_macros(dependent_macros, package="my_project")

# Render macro with dependency
rendered = context.render("SELECT {{ calculate_total() }} FROM sales")
print(rendered)  # Output: SELECT SUM(revenue) FROM sales

Built-in dbt Functions

from pathlib import Path
from sqlmesh.dbt.context import DbtContext

# DbtContext comes with built-in dbt functions pre-registered
context = DbtContext(
    project_root=Path("/dbt/project"),
    profile_name="analytics"
)
context.project_name = "my_project"

# Add model references for ref() function
from sqlmesh.dbt.model import ModelConfig
context.add_models({
    "stg_customers": ModelConfig(name="stg_customers", ...),
    "stg_orders": ModelConfig(name="stg_orders", ...)
})

# Add source references for source() function
from sqlmesh.dbt.source import SourceConfig
context.add_sources({
    "raw.customers": SourceConfig(name="customers", ...),
    "raw.orders": SourceConfig(name="orders", ...)
})

# Now built-in functions work in rendering
sql = """
SELECT
    c.customer_id,
    COUNT(o.order_id) as order_count
FROM {{ ref('stg_customers') }} c
LEFT JOIN {{ ref('stg_orders') }} o
    ON c.customer_id = o.customer_id
"""
rendered = context.render(sql)
print(rendered)
# Outputs SQL with ref() calls replaced by actual table references

Related Pages

Implements Principle

• Principle:TobikoData_Sqlmesh_Macro_And_Jinja_Translation

• Environment:TobikoData_Sqlmesh_Dbt_Compatibility