From e97a4275a17b18d6592f758e0a9f48efd4d61b10 Mon Sep 17 00:00:00 2001
From: Dominic Tarro <57306102+dominictarro@users.noreply.github.com>
Date: Mon, 27 Jan 2025 16:58:57 -0500
Subject: [PATCH 1/4] fix: check the type of dbt result when converting to
 markdown

trigger_dbt_cli_command will respond with more than one type depending on the command invoked. This change uses the appropriate var depending on that result type.
---
 .../prefect-dbt/prefect_dbt/cli/commands.py            | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git a/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py b/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
index f7da28a31fe3..7ec8a09c7a92 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
@@ -5,6 +5,7 @@
 from typing import Any, Dict, List, Optional, Union
 
 import yaml
+from dbt.artifacts.schemas.run.v5.run import RunResultOutput, RunResult
 from dbt.cli.main import dbtRunner, dbtRunnerResult
 from dbt.contracts.results import ExecutionResult, NodeStatus
 from prefect_shell.commands import ShellOperation
@@ -864,9 +865,12 @@ def create_summary_markdown(run_results: dict, command: str) -> str:
         markdown += _create_unsuccessful_markdown(run_results=run_results)
 
     if run_results["Success"] != []:
-        successful_runs_str = "\n".join(
-            [f"* {r.node.name}" for r in run_results["Success"]]
-        )
+        successful_runs_str = ""
+        for r in run_results["Success"]:
+            if isinstance(r, RunResult):
+                successful_runs_str += f"* {r.node.name}\n"
+            elif isinstance(r, RunResultOutput):
+                successful_runs_str += f"* {r.relation_name}\n"
         markdown += f"""\n## Successful Nodes ✅\n\n{successful_runs_str}\n\n"""
 
     return markdown

From 0a0e50ef9d4e92ac5f30f0be55e91c36481e3726 Mon Sep 17 00:00:00 2001
From: Dominic Tarro <57306102+dominictarro@users.noreply.github.com>
Date: Mon, 27 Jan 2025 17:25:18 -0500
Subject: [PATCH 2/4] fix: changed to a better variable for discriminating the
 command
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

From a test run

```
## Successful Nodes ✅

* unique_id=macro.dbt_external_tables.stage_external_sources message=None relation_name=None adapter_response={}
```
---
 src/integrations/prefect-dbt/prefect_dbt/cli/commands.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py b/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
index 7ec8a09c7a92..cd74539f0095 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
@@ -870,7 +870,7 @@ def create_summary_markdown(run_results: dict, command: str) -> str:
             if isinstance(r, RunResult):
                 successful_runs_str += f"* {r.node.name}\n"
             elif isinstance(r, RunResultOutput):
-                successful_runs_str += f"* {r.relation_name}\n"
+                successful_runs_str += f"* {r.unique_id}\n"
         markdown += f"""\n## Successful Nodes ✅\n\n{successful_runs_str}\n\n"""
 
     return markdown

From 41bb54b3f23fe0454d2a65a1dcdaa1ee23c216f6 Mon Sep 17 00:00:00 2001
From: Dominic Tarro <57306102+dominictarro@users.noreply.github.com>
Date: Tue, 28 Jan 2025 09:30:51 -0500
Subject: [PATCH 3/4] refactor: swapped import order

---
 src/integrations/prefect-dbt/prefect_dbt/cli/commands.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py b/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
index cd74539f0095..9bc5ab2f5f2d 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
@@ -5,7 +5,7 @@
 from typing import Any, Dict, List, Optional, Union
 
 import yaml
-from dbt.artifacts.schemas.run.v5.run import RunResultOutput, RunResult
+from dbt.artifacts.schemas.run.v5.run import RunResult, RunResultOutput
 from dbt.cli.main import dbtRunner, dbtRunnerResult
 from dbt.contracts.results import ExecutionResult, NodeStatus
 from prefect_shell.commands import ShellOperation

From b8a900e914e04e74edf998bf98321fc150dbcb15 Mon Sep 17 00:00:00 2001
From: Dominic Tarro <57306102+dominictarro@users.noreply.github.com>
Date: Tue, 28 Jan 2025 15:35:17 -0500
Subject: [PATCH 4/4] fix: support runs that output a FreshnessResult

---
 src/integrations/prefect-dbt/prefect_dbt/cli/commands.py | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py b/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
index 9bc5ab2f5f2d..1e130a77024f 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
@@ -5,6 +5,7 @@
 from typing import Any, Dict, List, Optional, Union
 
 import yaml
+from dbt.artifacts.schemas.freshness.v3.freshness import FreshnessResult
 from dbt.artifacts.schemas.run.v5.run import RunResult, RunResultOutput
 from dbt.cli.main import dbtRunner, dbtRunnerResult
 from dbt.contracts.results import ExecutionResult, NodeStatus
@@ -867,10 +868,12 @@ def create_summary_markdown(run_results: dict, command: str) -> str:
     if run_results["Success"] != []:
         successful_runs_str = ""
         for r in run_results["Success"]:
-            if isinstance(r, RunResult):
+            if isinstance(r, (RunResult, FreshnessResult)):
                 successful_runs_str += f"* {r.node.name}\n"
             elif isinstance(r, RunResultOutput):
                 successful_runs_str += f"* {r.unique_id}\n"
+            else:
+                successful_runs_str += f"* {r}\n"
         markdown += f"""\n## Successful Nodes ✅\n\n{successful_runs_str}\n\n"""
 
     return markdown