[DBT] Support profile config field, target.target_name, and an additional variant of config field missing the "+" (#597)

Author: crericha

sqlmesh/dbt/common.py

@@ -54,6 +54,7 @@ class DbtContext:
     project_root: Path = Path()
     target_name: t.Optional[str] = None
     project_name: t.Optional[str] = None
+    profile_name: t.Optional[str] = None
     project_schema: t.Optional[str] = None
     jinja_macros: JinjaMacroRegistry = field(
         default_factory=lambda: JinjaMacroRegistry(create_builtins_module="sqlmesh.dbt")

sqlmesh/dbt/package.py

@@ -121,7 +121,7 @@ def load_config(
             nested_config = {}
             fields = {}
             for key, value in data.items():
-                if key.startswith("+") or key in config_fields:
+                if key.startswith("+") or key in config_fields or not isinstance(value, dict):
                     fields[key[1:]] = value
                 else:
                     nested_config[key] = value

sqlmesh/dbt/profile.py

@@ -43,13 +43,16 @@ def load(cls, context: DbtContext) -> Profile:
         Returns:
             The Profile for the specified project
         """
-        if not context.project_name:
+        if not context.profile_name:
             project_file = Path(context.project_root, PROJECT_FILENAME)
             if not project_file.exists():
                 raise ConfigError(f"Could not find {PROJECT_FILENAME} in {context.project_root}")
 
-            context.project_name = context.render(load_yaml(project_file).get("name", ""))
-            if not context.project_name:
+            project_yaml = load_yaml(project_file)
+            context.profile_name = context.render(
+                project_yaml.get("profile", "")
+            ) or context.render(project_yaml.get("name", ""))
+            if not context.profile_name:
                 raise ConfigError(f"{project_file.stem} must include project name.")
 
         profile_filepath = cls._find_profile(context.project_root)
@@ -80,19 +83,19 @@ def _read_profile(
             source = file.read()
         contents = load_yaml(context.render(source))
 
-        project_data = contents.get(context.project_name)
+        project_data = contents.get(context.profile_name)
         if not project_data:
-            raise ConfigError(f"Project '{context.project_name}' does not exist in profiles.")
+            raise ConfigError(f"Profile '{context.profile_name}' not found in profiles.")
 
         outputs = project_data.get("outputs")
         if not outputs:
-            raise ConfigError(f"No outputs exist in profiles for Project '{context.project_name}'.")
+            raise ConfigError(f"No outputs exist in profiles for '{context.profile_name}'.")
 
         targets = {name: TargetConfig.load(name, output) for name, output in outputs.items()}
         default_target = context.render(project_data.get("target"))
         if default_target not in targets:
             raise ConfigError(
-                f"Default target '{default_target}' not specified in profiles for Project '{context.project_name}'."
+                f"Default target '{default_target}' not specified in profiles for '{context.profile_name}'."
             )
 
         return (targets, default_target)

sqlmesh/dbt/project.py

@@ -56,6 +56,10 @@ def load(cls, context: DbtContext) -> Project:
         if not context.project_name:
             raise ConfigError(f"{project_file_path.stem} must include project name.")
 
+        context.profile_name = (
+            context.render(project_yaml.get("profile", "")) or context.project_name
+        )
+
         profile = Profile.load(context)
         context.target = (
             profile.targets[context.target_name]

sqlmesh/dbt/target.py

@@ -31,11 +31,9 @@ class TargetConfig(abc.ABC, PydanticModel):
         threads: The number of threads to run on
     """
 
-    # sqlmesh
-    name: str = ""
-
     # dbt
     type: str
+    name: str = ""
     schema_: str = Field(alias="schema")
     threads: int = 1
     profile_name: t.Optional[str] = None
@@ -76,6 +74,7 @@ def to_sqlmesh(self) -> ConnectionConfig:
     def target_jinja(self, profile_name: str) -> AttributeDict:
         fields = self.dict().copy()
         fields["profile_name"] = profile_name
+        fields["target_name"] = self.name
         return AttributeDict(fields)