sqlmesh.dbt.package

  1from __future__ import annotations
  2
  3import logging
  4import typing as t
  5from pathlib import Path
  6
  7from sqlmesh.dbt.common import PROJECT_FILENAME, Dependencies, load_yaml
  8from sqlmesh.dbt.model import ModelConfig
  9from sqlmesh.dbt.seed import SeedConfig
 10from sqlmesh.dbt.source import SourceConfig
 11from sqlmesh.dbt.test import TestConfig
 12from sqlmesh.utils.errors import ConfigError
 13from sqlmesh.utils.jinja import MacroInfo
 14from sqlmesh.utils.pydantic import PydanticModel
 15
 16if t.TYPE_CHECKING:
 17    from sqlmesh.dbt.context import DbtContext
 18
 19
 20logger = logging.getLogger(__name__)
 21
 22
 23class MacroConfig(PydanticModel):
 24    """Class to contain macro configuration"""
 25
 26    info: MacroInfo
 27    dependencies: Dependencies
 28    path: Path
 29
 30
 31class HookConfig(PydanticModel):
 32    """Class to contain on run start / on run end hooks."""
 33
 34    sql: str
 35    index: int
 36    path: Path
 37    dependencies: Dependencies
 38
 39
 40class MaterializationConfig(PydanticModel):
 41    """Class to contain custom materialization configuration."""
 42
 43    name: str
 44    adapter: str
 45    definition: str
 46    dependencies: Dependencies
 47    path: Path
 48
 49
 50class Package(PydanticModel):
 51    """Class to contain package configuration"""
 52
 53    name: str
 54    tests: t.Dict[str, TestConfig]
 55    sources: t.Dict[str, SourceConfig]
 56    seeds: t.Dict[str, SeedConfig]
 57    models: t.Dict[str, ModelConfig]
 58    variables: t.Dict[str, t.Any]
 59    macros: t.Dict[str, MacroConfig]
 60    materializations: t.Dict[str, MaterializationConfig]
 61    on_run_start: t.Dict[str, HookConfig]
 62    on_run_end: t.Dict[str, HookConfig]
 63    files: t.Set[Path]
 64
 65    @property
 66    def macro_infos(self) -> t.Dict[str, MacroInfo]:
 67        return {name: macro.info for name, macro in self.macros.items()}
 68
 69
 70class PackageLoader:
 71    """Loader for DBT packages"""
 72
 73    def __init__(self, context: DbtContext):
 74        self._context = context
 75
 76    def load(self, package_root: Path) -> Package:
 77        """
 78        Loads the specified package.
 79
 80        Returns:
 81            Package containing the configuration found within this package
 82        """
 83        logger.debug("Loading package at '%s'.", package_root)
 84        project_file_path = package_root / PROJECT_FILENAME
 85        logger.debug("Processing project file '%s'.", project_file_path)
 86        if not project_file_path.exists():
 87            raise ConfigError(f"Could not find {PROJECT_FILENAME} in '{package_root}'.")
 88
 89        project_yaml = load_yaml(project_file_path)
 90        package_name = self._context.render(project_yaml.get("name", ""))
 91        if not package_name:
 92            raise ConfigError(f"Package '{package_root}' must include package name.")
 93
 94        # Only include globally-scoped variables (i.e. filter out the package-scoped ones)
 95        logger.debug("Processing project variables.")
 96
 97        all_variables = project_yaml.get("vars") or {}
 98        all_variables.update(all_variables.pop(package_name, None) or {})
 99
100        package_variables = {
101            var: value for var, value in all_variables.items() if not isinstance(value, dict)
102        }
103
104        tests = _fix_paths(self._context.manifest.tests(package_name), package_root)
105        models = _fix_paths(self._context.manifest.models(package_name), package_root)
106        seeds = _fix_paths(self._context.manifest.seeds(package_name), package_root)
107        macros = _fix_paths(self._context.manifest.macros(package_name), package_root)
108        materializations = _fix_paths(self._context.manifest.materializations(), package_root)
109        on_run_start = _fix_paths(self._context.manifest.on_run_start(package_name), package_root)
110        on_run_end = _fix_paths(self._context.manifest.on_run_end(package_name), package_root)
111        sources = self._context.manifest.sources(package_name)
112
113        config_paths = {
114            config.path.absolute()  # type: ignore
115            for configs in [models.values(), seeds.values(), macros.values()]
116            for config in configs
117        }
118        for path in package_root.glob("**/*.y*ml"):
119            config_paths.add(path.absolute())
120
121        return Package(
122            name=package_name,
123            tests=tests,
124            models=models,
125            sources=sources,
126            seeds=seeds,
127            variables=package_variables,
128            macros=macros,
129            materializations=materializations,
130            files=config_paths,
131            on_run_start=on_run_start,
132            on_run_end=on_run_end,
133        )
134
135
136T = t.TypeVar(
137    "T", TestConfig, ModelConfig, MacroConfig, MaterializationConfig, SeedConfig, HookConfig
138)
139
140
141def _fix_paths(configs: t.Dict[str, T], package_root: Path) -> t.Dict[str, T]:
142    return {
143        name: config.copy(update={"path": package_root / config.path})
144        for name, config in configs.items()
145    }