Repository: PrefectHQ/prefect
Pull request: fix `create_markdown_artifact` usage in `prefect-dbt`
closes https://github.com/PrefectHQ/prefect/issues/16924

https://github.com/PrefectHQ/prefect/pull/16877 removed `sync_compatible` from artifact methods / top-level utils

Problem statement:
prefect-dbt trigger_dbt_cli_command fails with ‘TypeError: object UUID can’t be used in ‘await’ expression’ after minor version 3.1.15 was published
### Bug summary

We are using the prefect-dbt package to trigger dbt commands within a Prefect flow. Without making any changes to our environment, all runs started failing after a minor version release of Prefect that was published around 9 PM (UTC) on 30 Jan 2025 (version 3.1.15). The failure occurs with the following error message:
```
TypeError: object UUID can't be used in 'await' expression
03:31:08 PM
trigger_dbt_cli_command-4e8
prefect.task_runs
Finished in state Failed("Task run encountered an exception TypeError: object UUID can't be used in 'await' expression")
```

This issue seems to be caused by the create_markdown_artifact() function inside the trigger_dbt_cli_command task.

### Full Stacktrace
```
Task run failed with exception: TypeError("object UUID can't be used in 'await' expression") - Retries are exhausted
Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/prefect/task_engine.py", line 805, in run_context
    yield self
  File "/usr/local/lib/python3.12/site-packages/prefect/task_engine.py", line 1387, in run_task_sync
    engine.call_task_fn(txn)
  File "/usr/local/lib/python3.12/site-packages/prefect/task_engine.py", line 828, in call_task_fn
    result = call_with_parameters(self.task.fn, parameters)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/prefect/utilities/callables.py", line 208, in call_with_parameters
    return fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/prefect/utilities/asyncutils.py", line 353, in coroutine_wrapper
    return run_coro_as_sync(ctx_call())
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/prefect/utilities/asyncutils.py", line 209, in run_coro_as_sync
    return call.result()
           ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/prefect/_internal/concurrency/calls.py", line 330, in result
    return self.future.result(timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/prefect/_internal/concurrency/calls.py", line 193, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.12/site-packages/prefect/_internal/concurrency/calls.py", line 403, in _run_async
    result = await coro
             ^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/prefect/utilities/asyncutils.py", line 190, in coroutine_wrapper
    return await task
           ^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/prefect/utilities/asyncutils.py", line 343, in ctx_call
    result = await async_fn(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/prefect_dbt/cli/commands.py", line 195, in trigger_dbt_cli_command
    artifact_id = await create_markdown_artifact(
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: object UUID can't be used in 'await' expression
```

### Steps to Reproduce
1.	Use the prefect-dbt package to run the `trigger_dbt_cli_command` with params:
```
from prefect import flow
from prefect_dbt.cli.commands import trigger_dbt_cli_command

result = trigger_dbt_cli_command(
        command='build',
        create_summary_artifact=True,
        summary_artifact_key='artifact_example'
    )
```
2.	Run the flow.
3.	The flow fails with TypeError: object UUID can't be used in 'await' expression.

### Expected Behavior

The Prefect task should complete successfully and, if create_summary_artifact=True, create a markdown artifact without error.

### Actual Behavior

The task fails due to an invalid await call on a UUID object.

### Possible Cause

The issue originates from the following line in prefect-dbt:
```
artifact_id = await create_markdown_artifact(
    markdown=markdown,
    key=summary_artifact_key,
)
```

Since create_markdown_artifact() appears to return a UUID instead of an awaitable coroutine in the latest Prefect release, this results in the TypeError.

This change was not introduced by an upgrade on my end, but all runs started failing after a minor Prefect version release around 9 PM last night (30 Jan) - version 3.1.15.

### Version info

```Text
Version:             3.1.15
API version:         0.8.4
Python version:      3.12.6
Git commit:          3ac3d548
Built:               Thu, Jan 30, 2025 11:31 AM
OS/Arch:             darwin/arm64
Profile:             ephemeral
Server type:         cloud
Pydantic version:    2.10.6
Integrations:
  prefect-gcp:       0.6.2
  prefect-aws:       0.5.3
  prefect-dbt:       0.6.4
  prefect-shell:     0.3.1
```

### Additional context

When the parameters create_summary_artifact and summary_artifact_key are not supplied, the function behaves as expected. This is what we have done to get our production pipeline up and running, however, it would be good to have artifacts back online

Hints from issue discussion:
thanks for the issue @dylan-delport - this is likely related to https://github.com/PrefectHQ/prefect/pull/16877

Commit message: fix markdown creation

Patch to review:
diff --git a/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py b/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
index 1e130a77024f..73edcaa52771 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
@@ -195,8 +195,7 @@ def _stream_output(event):
         run_results = consolidate_run_results(result)
         markdown = create_summary_markdown(run_results, command)
         artifact_id = await create_markdown_artifact(
-            markdown=markdown,
-            key=summary_artifact_key,
+            markdown=markdown, key=summary_artifact_key, _sync=False
         )
         if not artifact_id:
             logger.error(f"Summary Artifact was not created for dbt {command} task")
@@ -848,12 +847,13 @@ def dbt_test_flow():
     return results
 
 
-def create_summary_markdown(run_results: dict, command: str) -> str:
+def create_summary_markdown(run_results: dict[str, Any], command: str) -> str:
     """
     Creates a Prefect task artifact summarizing the results
     of the above predefined prefrect-dbt task.
     """
-    markdown = f"# dbt {command} Task Summary\n"
+    prefix = "dbt" if not command.startswith("dbt") else ""
+    markdown = f"# {prefix} {command} Task Summary\n"
     markdown += _create_node_summary_table_md(run_results=run_results)
 
     if (
diff --git a/src/integrations/prefect-dbt/tests/cli/test_commands.py b/src/integrations/prefect-dbt/tests/cli/test_commands.py
index 60381e0f7ede..b614f553b5fe 100644
--- a/src/integrations/prefect-dbt/tests/cli/test_commands.py
+++ b/src/integrations/prefect-dbt/tests/cli/test_commands.py
@@ -1,5 +1,6 @@
 import datetime
 from pathlib import Path
+from typing import Any
 from unittest.mock import MagicMock
 
 import pytest
@@ -216,7 +217,9 @@ async def mock_dbt_runner_ls_success():
 
 
 @pytest.fixture
-def dbt_runner_model_result(monkeypatch, mock_dbt_runner_model_success):
+def dbt_runner_model_result(
+    monkeypatch: pytest.MonkeyPatch, mock_dbt_runner_model_success: dbtRunnerResult
+) -> None:
     _mock_dbt_runner_invoke_success = MagicMock(
         return_value=mock_dbt_runner_model_success
     )
@@ -226,13 +229,17 @@ def dbt_runner_model_result(monkeypatch, mock_dbt_runner_model_success):
 
 
 @pytest.fixture
-def dbt_runner_ls_result(monkeypatch, mock_dbt_runner_ls_success):
+def dbt_runner_ls_result(
+    monkeypatch: pytest.MonkeyPatch, mock_dbt_runner_ls_success: dbtRunnerResult
+) -> None:
     _mock_dbt_runner_ls_result = MagicMock(return_value=mock_dbt_runner_ls_success)
     monkeypatch.setattr("dbt.cli.main.dbtRunner.invoke", _mock_dbt_runner_ls_result)
 
 
 @pytest.fixture
-def dbt_runner_freshness_error(monkeypatch, mock_dbt_runner_freshness_error):
+def dbt_runner_freshness_error(
+    monkeypatch: pytest.MonkeyPatch, mock_dbt_runner_freshness_error: dbtRunnerResult
+) -> None:
     _mock_dbt_runner_freshness_error = MagicMock(
         return_value=mock_dbt_runner_freshness_error
     )
@@ -242,7 +249,9 @@ def dbt_runner_freshness_error(monkeypatch, mock_dbt_runner_freshness_error):
 
 
 @pytest.fixture
-def dbt_runner_freshness_success(monkeypatch, mock_dbt_runner_freshness_success):
+def dbt_runner_freshness_success(
+    monkeypatch: pytest.MonkeyPatch, mock_dbt_runner_freshness_success: dbtRunnerResult
+) -> None:
     _mock_dbt_runner_freshness_success = MagicMock(
         return_value=mock_dbt_runner_freshness_success
     )
@@ -253,7 +262,7 @@ def dbt_runner_freshness_success(monkeypatch, mock_dbt_runner_freshness_success)
 
 
 @pytest.fixture
-def dbt_runner_failed_result(monkeypatch):
+def dbt_runner_failed_result(monkeypatch: pytest.MonkeyPatch) -> None:
     _mock_dbt_runner_invoke_failed = MagicMock(
         return_value=dbtRunnerResult(
             success=False,
@@ -265,7 +274,7 @@ def dbt_runner_failed_result(monkeypatch):
 
 
 @pytest.fixture
-def profiles_dir(tmp_path):
+def profiles_dir(tmp_path: Path) -> str:
     return str(tmp_path) + "/.dbt"
 
 
@@ -558,6 +567,30 @@ def test_append_dirs_to_commands(
         )
 
 
+@pytest.mark.usefixtures("dbt_runner_freshness_success")
+def test_sync_dbt_cli_command_creates_artifact(
+    profiles_dir: str, dbt_cli_profile: Any
+) -> None:
+    @flow
+    def test_flow() -> None:
+        trigger_dbt_cli_command(
+            command="dbt source freshness",
+            profiles_dir=profiles_dir,
+            dbt_cli_profile=dbt_cli_profile,
+            summary_artifact_key="foo",
+            create_summary_artifact=True,
+        )
+
+    test_flow()
+    assert (a := Artifact.get(key="foo"))
+    assert a.type == "markdown"
+    assert isinstance(a.data, str) and a.data.startswith(
+        "#  dbt source freshness Task Summary"
+    )
+    assert "my_first_dbt_model" in a.data
+    assert "Successful Nodes" in a.data
+
+
 @pytest.mark.usefixtures("dbt_runner_model_result")
 async def test_run_dbt_build_creates_artifact(profiles_dir, dbt_cli_profile_bare):
     @flow
diff --git a/src/integrations/prefect-dbt/tests/conftest.py b/src/integrations/prefect-dbt/tests/conftest.py
index a42c0044df95..110ea77c7b30 100644
--- a/src/integrations/prefect-dbt/tests/conftest.py
+++ b/src/integrations/prefect-dbt/tests/conftest.py
@@ -86,7 +86,7 @@ def dbt_cli_profile():
 
 
 @pytest.fixture
-def dbt_cli_profile_bare():
+def dbt_cli_profile_bare() -> DbtCliProfile:
     target_configs = TargetConfigs(
         type="custom", schema="my_schema", extras={"account": "fake"}
     )


Code review comments:
diff_hunk: @@ -195,8 +195,7 @@ def _stream_output(event):
         run_results = consolidate_run_results(result)
         markdown = create_summary_markdown(run_results, command)
         artifact_id = await create_markdown_artifact(
-            markdown=markdown,
-            key=summary_artifact_key,
+            markdown=markdown, key=summary_artifact_key, _sync=False
path: src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
text: @author:
key change

@user1:
`_sync=False` feels like syntax we're _eventually_ going to move away from - wouldn't it be better to call `acreate_markdown_artifact` here?

@author:
yes it would, updated in [d4662fb](https://github.com/PrefectHQ/prefect/pull/16926/commits/d4662fbc95735ed8c212b3b1802cd1756591549c)