Add secondary_profiles to profile.py

.changes/unreleased/Under the Hood-20250214-123853.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Add secondary profiles to profile.py
+time: 2025-02-14T12:38:53.964266Z
+custom:
+    Author: aranke
+    Issue: XPLAT-241

core/dbt/config/profile.py

@@ -2,6 +2,8 @@
 from dataclasses import dataclass
 from typing import Any, Dict, Optional, Tuple
 
+from typing_extensions import Self
+
 from dbt.adapters.contracts.connection import Credentials, HasCredentials
 from dbt.clients.yaml_helper import load_yaml_text
 from dbt.contracts.project import ProfileConfig
@@ -60,6 +62,7 @@ class Profile(HasCredentials):
     credentials: Credentials
     profile_env_vars: Dict[str, Any]
     log_cache_events: bool
+    secondary_profiles: Dict[str, Self]
 
     def __init__(
         self,
@@ -79,6 +82,7 @@ def __init__(
         self.log_cache_events = (
             get_flags().LOG_CACHE_EVENTS
         )  # never available on init, set for adapter instantiation via AdapterRequiredConfig
+        self.secondary_profiles = {}
 
     def to_profile_info(self, serialize_credentials: bool = False) -> Dict[str, Any]:
         """Unlike to_project_config, this dict is not a mirror of any existing
@@ -229,7 +233,7 @@ def from_credentials(
         threads: int,
         profile_name: str,
         target_name: str,
-    ) -> "Profile":
+    ) -> Self:
         """Create a profile from an existing set of Credentials and the
         remaining information.
 
@@ -257,6 +261,7 @@ def render_profile(
         profile_name: str,
         target_override: Optional[str],
         renderer: ProfileRenderer,
+        is_secondary: bool = False,
     ) -> Tuple[str, Dict[str, Any]]:
         """This is a containment zone for the hateful way we're rendering
         profiles.
@@ -273,6 +278,11 @@ def render_profile(
         elif "target" in raw_profile:
             # render the target if it was parsed from yaml
             target_name = renderer.render_value(raw_profile["target"])
+        elif is_secondary and len(raw_profile["outputs"]) == 1:
+            # if we only have one target, we can infer the target name
+            # currently, this is only used for secondary profiles
+            target_name = next(iter(raw_profile["outputs"]))
+            fire_event(MissingProfileTarget(profile_name=profile_name, target_name=target_name))
         else:
             target_name = "default"
             fire_event(MissingProfileTarget(profile_name=profile_name, target_name=target_name))
@@ -293,7 +303,8 @@ def from_raw_profile_info(
         renderer: ProfileRenderer,
         target_override: Optional[str] = None,
         threads_override: Optional[int] = None,
-    ) -> "Profile":
+        is_secondary: bool = False,
+    ) -> Self:
         """Create a profile from its raw profile information.
 
          (this is an intermediate step, mostly useful for unit testing)
@@ -312,9 +323,14 @@ def from_raw_profile_info(
         """
         # TODO: should it be, and the values coerced to bool?
         target_name, profile_data = cls.render_profile(
-            raw_profile, profile_name, target_override, renderer
+            raw_profile, profile_name, target_override, renderer, is_secondary=is_secondary
         )
 
+        if is_secondary and "secondary_profiles" in profile_data:
+            raise DbtProfileError(
+                f"Secondary profile '{profile_name}' cannot have nested secondary profiles"
+            )
+
         # valid connections never include the number of threads, but it's
         # stored on a per-connection level in the raw configs
         threads = profile_data.pop("threads", DEFAULT_THREADS)
@@ -325,13 +341,31 @@ def from_raw_profile_info(
             profile_data, profile_name, target_name
         )
 
-        return cls.from_credentials(
+        profile = cls.from_credentials(
             credentials=credentials,
             profile_name=profile_name,
             target_name=target_name,
             threads=threads,
         )
 
+        for p in profile_data.pop("secondary_profiles", []):
+            for secondary_profile_name, secondary_raw_profile in p.items():
+                if secondary_profile_name in profile.secondary_profiles:
+                    raise DbtProfileError(
+                        f"Secondary profile '{secondary_profile_name}' is already defined"
+                    )
+
+                profile.secondary_profiles[secondary_profile_name] = cls.from_raw_profile_info(
+                    secondary_raw_profile,
+                    secondary_profile_name,
+                    renderer,
+                    target_override=target_override,
+                    threads_override=threads_override,
+                    is_secondary=True,
+                )
+
+        return profile
+
     @classmethod
     def from_raw_profiles(
         cls,
@@ -340,7 +374,7 @@ def from_raw_profiles(
         renderer: ProfileRenderer,
         target_override: Optional[str] = None,
         threads_override: Optional[int] = None,
-    ) -> "Profile":
+    ) -> Self:
         """
         :param raw_profiles: The profile data, from disk as yaml.
         :param profile_name: The profile name to use.
@@ -381,7 +415,7 @@ def render(
         profile_name_override: Optional[str] = None,
         target_override: Optional[str] = None,
         threads_override: Optional[int] = None,
-    ) -> "Profile":
+    ) -> Self:
         """Given the raw profiles as read from disk and the name of the desired
         profile if specified, return the profile component of the runtime
         config.

core/dbt/config/runtime.py

@@ -188,6 +188,7 @@ def from_parts(
             project_env_vars=project.project_env_vars,
             restrict_access=project.restrict_access,
             profile_env_vars=profile.profile_env_vars,
+            secondary_profiles=profile.secondary_profiles,
             profile_name=profile.profile_name,
             target_name=profile.target_name,
             threads=profile.threads,