Hello @Marvin With prefect3, it seems that a flow is not stopping at first task failure. Here is the code within my flow :

@task(name="DBT - Command")
        async def dbt_command(command):
            return await trigger_dbt_cli_command(
                command=command,
                overwrite_profiles=True,
                dbt_cli_profile=dbt_cli_profile,
                project_dir=project_dir_str,
            )

        # Install dbt dependencies
        await dbt_command("dbt deps")

        # Source freshness test
        await dbt_command(
            f"dbt source freshness --exclude {' '.join(exclude_sources)}"
        )

How to make it stop at first task failure ?

To make your flow stop at the first task failure in Prefect 3, you'll need to handle task failures explicitly, as Prefect by default may not stop the entire flow execution on individual task failures. Prefect allows customization for such behavior through retries, failure hooks, or by using flow control logic. Here's a general approach for your situation: 1. Error Management: Implement error handling within your tasks or flow. Use try-except blocks where you can capture exceptions and decide how to handle them. For instance, you can raise an exception to stop the flow when a task fails. 2. Custom Task Execution Logic: Prefect tasks can be configured with retry attempts and other failure strategies. You may need to check the documentation for "graceful failure" mechanisms or alerts that can notify upon task failure. This is touched upon here. Implementing a simple error check in your flow might look something like:

python
@task(name="Handle DBT Command Failure")
async def handle_dbt_command_failure(command):
    try:
        return await trigger_dbt_cli_command(
            command=command,
            overwrite_profiles=True,
            dbt_cli_profile=dbt_cli_profile,
            project_dir=project_dir_str,
        )
    except Exception as e:
        # Log the error and re-raise to stop the flow
        print(f"Task failed with error: {str(e)}")
        raise

# Install dbt dependencies
await handle_dbt_command_failure("dbt deps")

# Source freshness test
await handle_dbt_command_failure(
    f"dbt source freshness --exclude {' '.join(exclude_sources)}"
)

This is a simple illustrative solution. You can further explore Prefect's 3.x Transaction documentation to better suit your needs, available here. If you start talking to your flow, make sure it doesn’t reply... you might just have invented true AI!