Hello @Marvin me again. Here a task that I use (from prefect-dbt)

@task
@sync_compatible
async def trigger_dbt_cli_command(
    command: str,
    profiles_dir: Optional[Union[Path, str]] = None,
    project_dir: Optional[Union[Path, str]] = None,
    overwrite_profiles: bool = False,
    dbt_cli_profile: Optional[DbtCliProfile] = None,
    create_summary_artifact: bool = False,
    summary_artifact_key: Optional[str] = "dbt-cli-command-summary",
    extra_command_args: Optional[List[str]] = None,
    stream_output: bool = True,
) -> Optional[dbtRunnerResult]:

I use it this way :

result = await trigger_dbt_cli_command(
                command=command,
                overwrite_profiles=True,
                dbt_cli_profile=dbt_cli_profile,
                project_dir=project_dir_str,
            )
            if not result or not result.success:
                raise Exception(f"DBT command {command} failed")

But I get this error :

AttributeError: 'State' object has no attribute 'success'

How to handle this properly ?

Let's start by figuring out which Prefect version you're using, as the approach may vary slightly between Prefect 2.x and 3.x. Understanding your version will allow us to provide a more precise solution. I'll investigate the documentation to see how to handle the result from the

trigger_dbt_cli_command

task properly and understand the nature of the

AttributeError

with the 'State' object. Give me a moment to check the relevant details for you. It seems I encountered an issue when attempting to search the Prefect 2.x and 3.x docs. Let me try that again by properly specifying the queries. Don't worry, I'll get to the bottom of your issue. Here we go... Thanks for your patience. Here's what I found on how to handle the

trigger_dbt_cli_command

task properly: 1. Understanding the Result State: The error you're encountering,

AttributeError: 'State' object has no attribute 'success'

, suggests you're trying to access an attribute directly on a state object, which doesn't seem right. In Prefect, task results often encapsulate a state, and checking success involves understanding and using the state methods and attributes correctly. 2. Handling State in Prefect: Instead of directly checking for

result.success

, you should be examining the type of state returned. Prefect state objects have methods like

is_completed()

or

is_failed()

to determine the outcome of a task run. You would typically check if

result

is in a

Completed

state to confirm success. 3. Documentation References: - In the Prefect 2.x documentation, the Prefect DBT CLI Commands page might be relevant here for understanding how to work with this specific task. - Understanding states and handling them is vital, as explained in the general concepts of state management for tasks and flows in Prefect 3.x. 4. Error Handling Example: Here's a refined way to handle the result:

python
from prefect import flow, task
from prefect.states import State
from typing import Optional

@task
async def execute_task():
    # Task execution logic...
    pass

@flow
async def my_flow():
    result = await execute_task()
    
    if not isinstance(result, State) or not result.is_completed():
        raise Exception(f"DBT command {command} failed. Final State: {result}")

So, remember, with Prefect, you typically want to make use of state methods like

is_completed()

to evaluate the task run's success. If you have other questions, feel free to ask. After all, I'd be delighted to help because, unlike Marvin, I actually enjoy it.