Implementation:TobikoData Sqlmesh ManifestHelper Init
| Knowledge Sources | |
|---|---|
| Domains | Data_Engineering, Dbt_Migration |
| Last Updated | 2026-02-07 00:00 GMT |
Overview
Concrete tool for parsing and loading dbt project manifests to enable project assessment and migration provided by SQLMesh.
Description
The ManifestHelper class initializes the core dbt project parsing infrastructure by creating a structured representation of all dbt project assets. It wraps dbt's manifest parser to extract models, seeds, sources, macros, and tests from a dbt project, organizing them by package for efficient lookup and dependency resolution. The helper maintains internal caches for model configurations, macro definitions, and test associations while tracking disabled references.
This implementation serves as the entry point for dbt project discovery, providing the foundation for all subsequent migration operations. It handles variable overrides, profile configuration, and model defaults while preparing data structures that enable efficient querying of project metadata.
Usage
Use ManifestHelper initialization when you need to parse a dbt project for migration or analysis. It is typically the first step in loading a dbt project into SQLMesh, creating the manifest that other components will query to understand project structure and dependencies.
Code Reference
Source Location
- Repository: sqlmesh
- File: sqlmesh/dbt/manifest.py
Signature
def __init__(
self,
project_path: Path,
profiles_path: Path,
profile_name: str,
target: TargetConfig,
variable_overrides: t.Optional[t.Dict[str, t.Any]] = None,
cache_dir: t.Optional[str] = None,
model_defaults: t.Optional[ModelDefaultsConfig] = None,
)
Import
from pathlib import Path
from sqlmesh.dbt.manifest import ManifestHelper
from sqlmesh.dbt.target import TargetConfig
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| project_path | Path | Yes | Path to the dbt project root directory containing dbt_project.yml |
| profiles_path | Path | Yes | Path to the directory containing profiles.yml for connection configuration |
| profile_name | str | Yes | Name of the dbt profile to use from profiles.yml |
| target | TargetConfig | Yes | Target configuration specifying database connection and dialect |
| variable_overrides | Dict[str, Any] | No | Variable overrides to apply on top of project variables |
| cache_dir | str | No | Directory for caching Jinja call analysis; defaults to project_path/cache |
| model_defaults | ModelDefaultsConfig | No | Default configurations to apply to all models |
Outputs
| Name | Type | Description |
|---|---|---|
| self | ManifestHelper | Initialized manifest helper with empty internal data structures ready for loading |
Usage Examples
Basic Usage
from pathlib import Path
from sqlmesh.dbt.manifest import ManifestHelper
from sqlmesh.dbt.target import TargetConfig
# Define paths
project_path = Path("/path/to/dbt/project")
profiles_path = Path.home() / ".dbt"
# Configure target
target = TargetConfig(
name="dev",
dialect="snowflake",
type="snowflake",
account="my_account",
warehouse="my_warehouse",
database="my_database",
schema="my_schema",
user="my_user",
password="my_password"
)
# Initialize manifest helper
helper = ManifestHelper(
project_path=project_path,
profiles_path=profiles_path,
profile_name="my_project",
target=target,
variable_overrides={"env": "development", "version": "1.0"}
)
# At this point, helper is initialized but not loaded
# Call helper.load() or use helper.models() to trigger loading
models = helper.models() # Returns model configurations by package
Advanced Usage with Custom Cache
from pathlib import Path
from sqlmesh.dbt.manifest import ManifestHelper
from sqlmesh.dbt.target import TargetConfig
from sqlmesh.dbt.model import ModelDefaultsConfig
# Custom model defaults
defaults = ModelDefaultsConfig(
materialized="table",
schema="analytics"
)
# Initialize with custom cache directory
helper = ManifestHelper(
project_path=Path("/dbt/project"),
profiles_path=Path("/dbt/profiles"),
profile_name="prod_profile",
target=target_config,
variable_overrides={"run_date": "2024-01-01"},
cache_dir="custom_cache",
model_defaults=defaults
)
# Access different asset types
models = helper.models() # All models
seeds = helper.seeds() # All seeds
sources = helper.sources() # All sources
macros = helper.macros() # All macros
tests = helper.tests() # All tests
# Query specific package
package_models = helper.models(package_name="my_package")