sqlmesh.dbt.profile

  1from __future__ import annotations
  2
  3import logging
  4import os
  5import typing as t
  6from pathlib import Path
  7
  8from sqlmesh.dbt.common import PROJECT_FILENAME, load_yaml
  9from sqlmesh.dbt.context import DbtContext
 10from sqlmesh.dbt.target import TargetConfig
 11from sqlmesh.utils import yaml
 12from sqlmesh.utils.errors import ConfigError
 13
 14logger = logging.getLogger(__name__)
 15
 16
 17class Profile:
 18    """
 19    A class to read DBT profiles and obtain the project's target data warehouse configuration
 20    """
 21
 22    PROFILE_FILE = "profiles.yml"
 23
 24    def __init__(
 25        self,
 26        path: Path,
 27        target_name: str,
 28        target: TargetConfig,
 29    ):
 30        """
 31        Args:
 32            path: Path to the profile file
 33            target_name: Name of the target loaded
 34            target: TargetConfig for target_name
 35        """
 36        self.path = path
 37        self.target_name = target_name
 38        self.target = target
 39
 40    @classmethod
 41    def load(cls, context: DbtContext, target_name: t.Optional[str] = None) -> Profile:
 42        """
 43        Loads the profile for the specified project
 44
 45        Args:
 46            context: DBT context for this profile
 47
 48        Returns:
 49            The Profile for the specified project
 50        """
 51        if not context.profile_name:
 52            project_file = Path(context.project_root, PROJECT_FILENAME)
 53            if not project_file.exists():
 54                raise ConfigError(f"Could not find {PROJECT_FILENAME} in {context.project_root}")
 55
 56            project_yaml = load_yaml(project_file)
 57            context.profile_name = context.render(
 58                project_yaml.get("profile", "")
 59            ) or context.render(project_yaml.get("name", ""))
 60            if not context.profile_name:
 61                raise ConfigError(f"{project_file.stem} must include project name.")
 62
 63        profile_filepath = cls._find_profile(context.project_root, context.profiles_dir)
 64        if not profile_filepath:
 65            raise ConfigError(f"{cls.PROFILE_FILE} not found.")
 66
 67        target_name, target = cls._read_profile(profile_filepath, context, target_name)
 68        return Profile(profile_filepath, target_name, target)
 69
 70    @classmethod
 71    def _find_profile(cls, project_root: Path, profiles_dir: t.Optional[Path]) -> t.Optional[Path]:
 72        dir = os.environ.get("DBT_PROFILES_DIR", profiles_dir or "")
 73        path = Path(project_root, dir, cls.PROFILE_FILE)
 74        if path.exists():
 75            return path
 76        if dir:
 77            return None
 78
 79        path = Path(Path.home(), ".dbt", cls.PROFILE_FILE)
 80        if path.exists():
 81            return path
 82
 83        return None
 84
 85    @classmethod
 86    def _read_profile(
 87        cls, path: Path, context: DbtContext, target_name: t.Optional[str] = None
 88    ) -> t.Tuple[str, TargetConfig]:
 89        logger.debug("Processing profile '%s'.", path)
 90        project_data = load_yaml(path).get(context.profile_name)
 91        if not project_data:
 92            raise ConfigError(f"Profile '{context.profile_name}' not found in profiles.")
 93
 94        outputs = project_data.get("outputs")
 95        if not outputs:
 96            raise ConfigError(f"No outputs exist in profiles for '{context.profile_name}'.")
 97
 98        if not target_name:
 99            if "target" not in project_data:
100                raise ConfigError(f"No target specified for '{context.profile_name}'.")
101            target_name = context.render(project_data.get("target"))
102
103        if target_name not in outputs:
104            target_names = "\n".join(f"- {name}" for name in outputs)
105            raise ConfigError(
106                f"Target '{target_name}' not specified in profiles for '{context.profile_name}'. "
107                f"The valid target names for this profile are:\n{target_names}"
108            )
109
110        target_fields = load_yaml(context.render(yaml.dump(outputs[target_name])))
111        target = TargetConfig.load(
112            {"name": target_name, "profile_name": context.profile_name, **target_fields}
113        )
114
115        return (target_name, target)