I understand your issue - this is a known behavior with

trigger_dbt_cli_command

where the task can show as "completed" even when the underlying dbt command fails. This happens because by default, the command doesn't automatically translate dbt's exit codes into Prefect task states. Here's how you can properly handle dbt failures: 1. Use the

raise_on_error

parameter when calling `trigger_dbt_cli_command`:

python
from prefect_dbt.cli.commands import trigger_dbt_cli_command

@flow
def my_dbt_flow():
    # This will now raise an exception if dbt fails
    result = trigger_dbt_cli_command(
        command="run",
        project_dir="path/to/dbt/project",
        raise_on_error=True  # Add this parameter
    )

When

raise_on_error=True

, the task will raise a

DbtCliError

if the dbt command fails, which will properly mark your Prefect run as failed. 2. Alternatively, you can check the return code manually:

python
from prefect import flow
from prefect_dbt.cli.commands import trigger_dbt_cli_command

@flow
def my_dbt_flow():
    result = trigger_dbt_cli_command(
        command="run",
        project_dir="path/to/dbt/project"
    )
    
    if result.return_code != 0:
        raise Exception(f"dbt command failed with return code {result.return_code}")

The second approach gives you more control over how you want to handle different types of failures, as you can inspect both the return code and the output. Both methods will ensure that your Prefect run is properly marked as failed when dbt encounters an error, making it easier to track and log failures. Note: If you're using any logging systems or notification systems with Prefect, they will now properly trigger on these failures since the flow will actually fail instead of completing with a hidden dbt error. Would you like me to provide more details about any specific aspect of this error handling approach?