[dagster-dbt] Support profiles_dir and profile in DbtProject (#27416)

## Summary & Motivation

Same as #21108 but for `profiles_dir` and `profile`.

In response to PR #26928, fixes issue #26504

## How I Tested These Changes

Additional tests with BK

## Changelog

[dagster-dbt] Specifying a dbt profiles directory and profile is now
supported in `DbtProject`.

python_modules/libraries/dagster-dbt/dagster_dbt/core/resource.py

@@ -217,6 +217,12 @@ def __init__(
             if not state_path and project_dir.state_path:
                 state_path = project_dir.state_path
 
+            if not profiles_dir and project_dir.profiles_dir:
+                profiles_dir = project_dir.profiles_dir
+
+            if not profile and project_dir.profile:
+                profile = project_dir.profile
+
             if not target and project_dir.target:
                 target = project_dir.target
 

python_modules/libraries/dagster-dbt/dagster_dbt/dbt_project.py

@@ -11,6 +11,7 @@
 
 from dagster_dbt.errors import (
     DagsterDbtManifestNotFoundError,
+    DagsterDbtProfilesDirectoryNotFoundError,
     DagsterDbtProjectNotFoundError,
     DagsterDbtProjectYmlFileNotFoundError,
 )
@@ -153,6 +154,11 @@ class DbtProject(IHaveNew):
             The path, relative to the project directory, to output artifacts.
             It corresponds to the target path in dbt.
             Default: "target"
+        profiles_dir (Union[str, Path]):
+            The path to the directory containing your dbt `profiles.yml`.
+            By default, the current working directory is used, which is the dbt project directory.
+        profile (Optional[str]):
+            The profile from your dbt `profiles.yml` to use for execution, if it should be explicitly set.
         target (Optional[str]):
             The target from your dbt `profiles.yml` to use for execution, if it should be explicitly set.
         packaged_project_dir (Optional[Union[str, Path]]):
@@ -203,6 +209,8 @@ def get_env():
     name: str
     project_dir: Path
     target_path: Path
+    profiles_dir: Path
+    profile: Optional[str]
     target: Optional[str]
     manifest_path: Path
     packaged_project_dir: Optional[Path]
@@ -215,6 +223,8 @@ def __new__(
         project_dir: Union[Path, str],
         *,
         target_path: Union[Path, str] = Path("target"),
+        profiles_dir: Optional[Union[Path, str]] = None,
+        profile: Optional[str] = None,
         target: Optional[str] = None,
         packaged_project_dir: Optional[Union[Path, str]] = None,
         state_path: Optional[Union[Path, str]] = None,
@@ -223,6 +233,12 @@ def __new__(
         if not project_dir.exists():
             raise DagsterDbtProjectNotFoundError(f"project_dir {project_dir} does not exist.")
 
+        profiles_dir = Path(profiles_dir) if profiles_dir else project_dir
+        if not profiles_dir.exists():
+            raise DagsterDbtProfilesDirectoryNotFoundError(
+                f"profiles {profiles_dir} does not exist."
+            )
+
         packaged_project_dir = Path(packaged_project_dir) if packaged_project_dir else None
         if not using_dagster_dev() and packaged_project_dir and packaged_project_dir.exists():
             project_dir = packaged_project_dir
@@ -255,6 +271,8 @@ def __new__(
             name=dbt_project_yml["name"],
             project_dir=project_dir,
             target_path=target_path,
+            profiles_dir=profiles_dir,
+            profile=profile,
             target=target,
             manifest_path=manifest_path,
             state_path=project_dir.joinpath(state_path) if state_path else None,

python_modules/libraries/dagster-dbt/dagster_dbt/errors.py

@@ -19,6 +19,10 @@ class DagsterDbtProjectNotFoundError(DagsterDbtError):
     """Error when the specified project directory can not be found."""
 
 
+class DagsterDbtProfilesDirectoryNotFoundError(DagsterDbtError):
+    """Error when the specified profiles directory can not be found."""
+
+
 class DagsterDbtManifestNotFoundError(DagsterDbtError):
     """Error when we expect manifest.json to generated already but it is absent."""
 

python_modules/libraries/dagster-dbt/dagster_dbt_tests/core/test_resource.py

@@ -16,7 +16,7 @@
 from dagster_dbt.core.resource import DbtCliResource
 from dagster_dbt.dagster_dbt_translator import DagsterDbtTranslator, DagsterDbtTranslatorSettings
 from dagster_dbt.dbt_project import DbtProject
-from dagster_dbt.errors import DagsterDbtCliRuntimeError
+from dagster_dbt.errors import DagsterDbtCliRuntimeError, DagsterDbtProfilesDirectoryNotFoundError
 from dbt.version import __version__ as dbt_version
 from packaging import version
 from pydantic import ValidationError
@@ -232,6 +232,26 @@ def test_dbt_profile_configuration() -> None:
     ]
     assert dbt_cli_invocation.is_successful()
 
+    dbt_cli_invocation = (
+        DbtCliResource(
+            project_dir=DbtProject(
+                os.fspath(test_jaffle_shop_path), profile="jaffle_shop", target="dev"
+            )
+        )
+        .cli(["parse"])
+        .wait()
+    )
+
+    assert dbt_cli_invocation.process.args == [
+        "dbt",
+        "parse",
+        "--profile",
+        "jaffle_shop",
+        "--target",
+        "dev",
+    ]
+    assert dbt_cli_invocation.is_successful()
+
 
 @pytest.mark.parametrize(
     "profiles_dir", [None, test_jaffle_shop_path, os.fspath(test_jaffle_shop_path)]
@@ -246,17 +266,39 @@ def test_dbt_profiles_dir_configuration(profiles_dir: Union[str, Path]) -> None:
         .is_successful()
     )
 
+    assert (
+        DbtCliResource(
+            project_dir=DbtProject(os.fspath(test_jaffle_shop_path), profiles_dir=profiles_dir)
+        )
+        .cli(["parse"])
+        .is_successful()
+    )
+
     # profiles directory must exist
     with pytest.raises(ValidationError, match="does not exist"):
         DbtCliResource(project_dir=os.fspath(test_jaffle_shop_path), profiles_dir="nonexistent")
 
+    # Error is raised at the DbtProject level when a nonexistent directory is passed to a DbtProject object
+    with pytest.raises(DagsterDbtProfilesDirectoryNotFoundError, match="does not exist"):
+        DbtCliResource(
+            project_dir=DbtProject(os.fspath(test_jaffle_shop_path), profiles_dir="nonexistent")
+        )
+
     # profiles directory must contain profile configuration
     with pytest.raises(ValidationError, match="specify a valid path to a dbt profile directory"):
         DbtCliResource(
             project_dir=os.fspath(test_jaffle_shop_path),
             profiles_dir=f"{os.fspath(test_jaffle_shop_path)}/models",
         )
 
+    with pytest.raises(ValidationError, match="specify a valid path to a dbt profile directory"):
+        DbtCliResource(
+            project_dir=DbtProject(
+                os.fspath(test_jaffle_shop_path),
+                profiles_dir=f"{os.fspath(test_jaffle_shop_path)}/models",
+            )
+        )
+
 
 def test_dbt_project_dir_conflicting_env_var(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("DBT_PROJECT_DIR", "nonexistent")