[dagster-dbt] add dbt cloud logs to stdout after command runs (#32941)

## Summary & Motivation

Our dbt core integration will show logs to stdout for the dbt command
that was run, but we don't do that for dbt cloud. Per the dbt cloud
docs, it does not have a streaming log API endpoint, but you can get the
logs after the dbt commands complete.


Example output:

<img width="1290" height="858" alt="image"
src="https://github.com/user-attachments/assets/a0ffef97-a353-48ec-8156-0b68ec435927"
/>


## How I Tested These Changes

BK + pytest (locally) + our dbt cloud test environment


## Changelog

* [dagster-dbt] Add logs for dbt cloud to the stdout after the run
completes in dbt cloud

Author: cnolanminich

python_modules/libraries/dagster-dbt/dagster_dbt/cloud_v2/cli_invocation.py

@@ -1,3 +1,4 @@
+import sys
 from collections.abc import Iterator, Mapping, Sequence
 from typing import Any, Optional, Union
 
@@ -54,6 +55,12 @@ def wait(
         self, timeout: Optional[float] = None
     ) -> Iterator[Union[AssetCheckEvaluation, AssetCheckResult, AssetMaterialization, Output]]:
         run = self.run_handler.wait(timeout=timeout)
+
+        # Write dbt Cloud run logs to stdout
+        logs = self.run_handler.get_run_logs()
+        if logs:
+            sys.stdout.write(logs)
+
         if "run_results.json" in self.run_handler.list_run_artifacts():
             run_results = DbtCloudJobRunResults.from_run_results_json(
                 run_results_json=self.run_handler.get_run_results()

python_modules/libraries/dagster-dbt/dagster_dbt/cloud_v2/client.py

@@ -92,7 +92,7 @@ def _make_request(
         data: Optional[Mapping[str, Any]] = None,
         params: Optional[Mapping[str, Any]] = None,
         session_attr: str = "_get_session",
-    ) -> Mapping[str, Any]:
+    ) -> requests.Response:
         url = f"{base_url}/{endpoint}" if endpoint else base_url
 
         num_retries = 0
@@ -107,7 +107,7 @@ def _make_request(
                     timeout=self.request_timeout,
                 )
                 response.raise_for_status()
-                return response.json()
+                return response
             except RequestException as e:
                 self._log.error(
                     f"Request to dbt Cloud API failed for url {url} with method {method} : {e}"
@@ -143,7 +143,7 @@ def create_job(
         """
         if not description:
             description = "A job that runs dbt models, sources, and tests."
-        return self._make_request(
+        response = self._make_request(
             method="post",
             endpoint="jobs",
             base_url=self.api_v2_url,
@@ -155,7 +155,8 @@ def create_job(
                 "description": description,
                 "job_type": "other",
             },
-        )["data"]
+        )
+        return response.json()["data"]
 
     def list_jobs(
         self,
@@ -185,7 +186,7 @@ def list_jobs(
                 "limit": DAGSTER_DBT_CLOUD_LIST_JOBS_INDIVIDUAL_REQUEST_LIMIT,
                 "offset": len(results),
             },
-        )["data"]:
+        ).json()["data"]:
             results.extend(jobs)
             if len(jobs) < DAGSTER_DBT_CLOUD_LIST_JOBS_INDIVIDUAL_REQUEST_LIMIT:
                 break
@@ -201,7 +202,7 @@ def get_job_details(self, job_id: int) -> Mapping[str, Any]:
             method="get",
             endpoint=f"jobs/{job_id}",
             base_url=self.api_v2_url,
-        )["data"]
+        ).json()["data"]
 
     def destroy_job(self, job_id: int) -> Mapping[str, Any]:
         """Destroys a given dbt Cloud job.
@@ -213,7 +214,7 @@ def destroy_job(self, job_id: int) -> Mapping[str, Any]:
             method="delete",
             endpoint=f"jobs/{job_id}",
             base_url=self.api_v2_url,
-        )["data"]
+        ).json()["data"]
 
     def trigger_job_run(
         self, job_id: int, steps_override: Optional[Sequence[str]] = None
@@ -237,7 +238,7 @@ def trigger_job_run(
             data={"steps_override": steps_override, "cause": DAGSTER_ADHOC_TRIGGER_CAUSE}
             if steps_override
             else {"cause": DAGSTER_ADHOC_TRIGGER_CAUSE},
-        )["data"]
+        ).json()["data"]
 
     def get_runs_batch(
         self,
@@ -278,26 +279,35 @@ def get_runs_batch(
                 "finished_at__range": f"""["{finished_at_lower_bound.isoformat()}", "{finished_at_upper_bound.isoformat()}"]""",
                 "order_by": "finished_at",
             },
-        )
+        ).json()
         data = cast("Sequence[Mapping[str, Any]]", resp["data"])
         total_count = resp["extra"]["pagination"]["total_count"]
         return data, total_count
 
-    def get_run_details(self, run_id: int) -> Mapping[str, Any]:
+    def get_run_details(
+        self, run_id: int, include_related: Optional[Sequence[str]] = None
+    ) -> Mapping[str, Any]:
         """Retrieves the details of a given dbt Cloud Run.
 
         Args:
-            run_id (str): The dbt Cloud Run ID. You can retrieve this value from the
+            run_id (int): The dbt Cloud Run ID. You can retrieve this value from the
                 URL of the given run in the dbt Cloud UI.
+            include_related (Optional[Sequence[str]]): List of related fields to pull with the run.
+                Valid values are "trigger", "job", "debug_logs", and "run_steps".
 
         Returns:
             Dict[str, Any]: Parsed json data representing the API response.
         """
+        params = {}
+        if include_related:
+            params["include_related"] = ",".join(include_related)
+
         return self._make_request(
             method="get",
             endpoint=f"runs/{run_id}",
             base_url=self.api_v2_url,
-        )["data"]
+            params=params,
+        ).json()["data"]
 
     def poll_run(
         self,
@@ -352,21 +362,25 @@ def list_run_artifacts(
                 endpoint=f"runs/{run_id}/artifacts",
                 base_url=self.api_v2_url,
                 session_attr="_get_artifact_session",
-            )["data"],
+            ).json()["data"],
         )
 
     def get_run_artifact(self, run_id: int, path: str) -> Mapping[str, Any]:
         """Retrieves an artifact at the given path for a given dbt Cloud Run.
 
+        Args:
+            run_id (int): The dbt Cloud Run ID.
+            path (str): The path to the artifact (e.g., "run_results.json", "manifest.json").
+
         Returns:
-            Dict[str, Any]: Parsed json data representing the API response.
+            Dict[str, Any]: Parsed json data representing the artifact.
         """
         return self._make_request(
             method="get",
             endpoint=f"runs/{run_id}/artifacts/{path}",
             base_url=self.api_v2_url,
             session_attr="_get_artifact_session",
-        )
+        ).json()
 
     def get_run_results_json(self, run_id: int) -> Mapping[str, Any]:
         """Retrieves the run_results.json artifact of a given dbt Cloud Run.
@@ -384,6 +398,65 @@ def get_run_manifest_json(self, run_id: int) -> Mapping[str, Any]:
         """
         return self.get_run_artifact(run_id=run_id, path="manifest.json")
 
+    def get_run_logs(self, run_id: int, max_retries: int = 3, retry_delay: float = 2.0) -> str:
+        """Retrieves the stdout/stderr logs from a given dbt Cloud Run.
+
+        This method fetches logs from the run_steps field by calling get_run_details
+        with include_related=["run_steps"]. Each step contains a logs field with
+        the stdout/stderr output for that step.
+
+        Note: There can be a slight delay between when a run completes and when the logs
+        are fully populated in the API. This method will retry a few times if it detects
+        completed steps with empty logs.
+
+        Args:
+            run_id (int): The dbt Cloud Run ID.
+            max_retries (int): Maximum number of times to retry fetching logs if empty. Defaults to 3.
+            retry_delay (float): Time in seconds to wait between retries. Defaults to 2.0.
+
+        Returns:
+            str: The concatenated log text content from all run steps.
+        """
+        for attempt in range(max_retries):
+            run_details = self.get_run_details(run_id=run_id, include_related=["run_steps"])
+
+            logs_parts = []
+            run_steps = run_details.get("run_steps", [])
+            completed_steps_with_empty_logs = 0
+
+            for step in run_steps:
+                step_name = step.get("name", "Unknown Step")
+                step_logs = step.get("logs", "")
+                step_status = step.get("status_humanized", "unknown")
+
+                # Track completed steps with empty logs
+                if step_status == "Success" and not step_logs:
+                    completed_steps_with_empty_logs += 1
+
+                if step_logs:
+                    logs_parts.append(f"=== Step: {step_name} ===")
+                    logs_parts.append(step_logs)
+                    logs_parts.append("")  # Empty line between steps
+
+            # If we have completed steps with empty logs and retries left, wait and try again
+            if completed_steps_with_empty_logs > 0 and attempt < max_retries - 1:
+                self._log.warning(
+                    f"Found {completed_steps_with_empty_logs} completed steps with empty logs for run {run_id}. "
+                    f"Retrying in {retry_delay} seconds..."
+                )
+                time.sleep(retry_delay)
+                continue
+
+            # Either we got all logs or we're out of retries
+            if completed_steps_with_empty_logs > 0:
+                self._log.warning(
+                    f"Still missing logs for {completed_steps_with_empty_logs} completed steps after {max_retries} attempts"
+                )
+
+            return "\n".join(logs_parts) if logs_parts else ""
+
+        return ""
+
     def get_project_details(self, project_id: int) -> Mapping[str, Any]:
         """Retrieves the details of a given dbt Cloud Project.
 
@@ -398,7 +471,7 @@ def get_project_details(self, project_id: int) -> Mapping[str, Any]:
             method="get",
             endpoint=f"projects/{project_id}",
             base_url=self.api_v2_url,
-        )["data"]
+        ).json()["data"]
 
     def get_environment_details(self, environment_id: int) -> Mapping[str, Any]:
         """Retrieves the details of a given dbt Cloud Environment.
@@ -414,7 +487,7 @@ def get_environment_details(self, environment_id: int) -> Mapping[str, Any]:
             method="get",
             endpoint=f"environments/{environment_id}",
             base_url=self.api_v2_url,
-        )["data"]
+        ).json()["data"]
 
     def get_account_details(self) -> Mapping[str, Any]:
         """Retrieves the details of the account associated to the dbt Cloud workspace.
@@ -426,7 +499,7 @@ def get_account_details(self) -> Mapping[str, Any]:
             method="get",
             endpoint=None,
             base_url=self.api_v2_url,
-        )["data"]
+        ).json()["data"]
 
     def verify_connection(self) -> None:
         """Verifies the connection to the dbt Cloud REST API."""

python_modules/libraries/dagster-dbt/dagster_dbt/cloud_v2/run_handler.py

@@ -13,6 +13,7 @@
 )
 from dagster._record import record
 from dateutil import parser
+from requests.exceptions import RequestException
 
 from dagster_dbt.asset_utils import build_dbt_specs, get_asset_check_key_for_test
 from dagster_dbt.cloud_v2.client import DbtCloudWorkspaceClient
@@ -61,6 +62,22 @@ def get_manifest(self) -> Mapping[str, Any]:
     def list_run_artifacts(self) -> Sequence[str]:
         return self.client.list_run_artifacts(run_id=self.run_id)
 
+    def get_run_logs(self) -> Optional[str]:
+        """Retrieves the stdout/stderr logs from the completed dbt Cloud run.
+
+        This method fetches logs from the run_steps by calling get_run_details
+        with include_related=["run_steps"].
+
+        Returns:
+            Optional[str]: The concatenated log text content from all run steps,
+                or None if logs are not available.
+        """
+        try:
+            return self.client.get_run_logs(run_id=self.run_id)
+        except RequestException as e:
+            logger.warning(f"Failed to retrieve logs for run {self.run_id}: {e}")
+            return None
+
 
 def get_completed_at_timestamp(result: Mapping[str, Any]) -> float:
     # result["timing"] is a list of events in run_results.json

python_modules/libraries/dagster-dbt/dagster_dbt_tests/cloud_v2/conftest.py

@@ -654,6 +654,45 @@ def get_sample_run_response(run_status: int, job_id: int = TEST_ADHOC_JOB_ID) ->
 )
 
 
+def get_sample_run_response_with_logs(
+    run_status: int, job_id: int = TEST_ADHOC_JOB_ID
+) -> Mapping[str, Any]:
+    """Returns a sample run response with realistic run_steps logs."""
+    return {
+        "data": {
+            **get_sample_run_data(run_status=run_status, job_id=job_id),
+            "run_steps": [
+                {
+                    "id": 1,
+                    "run_id": TEST_RUN_ID,
+                    "account_id": TEST_ACCOUNT_ID,
+                    "index": 1,
+                    "status": 10,
+                    "name": "Clone git repository",
+                    "logs": "Cloning into '/tmp/jobs/12345/target'...\nSuccessfully cloned repository.\nChecked out to abc123.",
+                    "debug_logs": "",
+                },
+                {
+                    "id": 2,
+                    "run_id": TEST_RUN_ID,
+                    "account_id": TEST_ACCOUNT_ID,
+                    "index": 2,
+                    "status": 10,
+                    "name": "Invoke dbt with `dbt build`",
+                    "logs": "15:49:32  Running dbt...\n15:49:34  Found 5 models, 13 data tests\n15:49:35  Completed successfully",
+                    "debug_logs": "",
+                },
+            ],
+        },
+        "status": {
+            "code": 200,
+            "is_success": True,
+            "user_message": "Success!",
+            "developer_message": "",
+        },
+    }
+
+
 # Taken from dbt Cloud REST API documentation
 # https://docs.getdbt.com/dbt-cloud/api-v2#/operations/List%20Runs
 def get_sample_list_runs_sample(data: Sequence[Mapping[str, Any]], count: int, total_count: int):