trigger_dbt_cloud_job_run_and_wait_for_completion in prefect-dbt doesn't work as expected in 3.0.1

[bjorhn]

Bug summary

In prefect 3.0.1, trigger_dbt_cloud_job_run_and_wait_for_completion appears to exit immediately. This happens when executing it both from a sync or an async flow. We usually run it from a sync flow but decided to try with async to see if it'd work, or if perhaps we had to await trigger_dbt_cloud_job_run_and_wait_for_completion, which is an async subflow (marked as sync compatible, I believe). Awaiting it produced an error message.

I don't have any detailed logs at the moment but it should be easy to re-create.

In our case, the parent flow that is attempting to trigger and wait for the dbt cloud job run simply exits successfully (state = Completed), without trigger_dbt_cloud_job_run_and_wait_for_completion having created a subflow.

dbt_cloud_logs = trigger_dbt_cloud_job_run_and_wait_for_completion(
            dbt_cloud_credentials=dbt_cloud_credentials,
            job_id=DBT_JOB_IDS.get(env.value),
            trigger_job_run_options=trigger_job_run_options,
            max_wait_seconds=10800,  # 3 hours
            retry_filtered_models_attempts=0,
        )

logger.info(dbt_cloud_logs)

We have a temporary workaround in place but we're not entirely sure why it's necessary:

dbt_cloud_logs = asyncio.run(
            trigger_dbt_cloud_job_run_and_wait_for_completion(
                dbt_cloud_credentials=dbt_cloud_credentials,
                job_id=DBT_JOB_IDS.get(env.value),
                trigger_job_run_options=trigger_job_run_options,
                max_wait_seconds=10800,  # 3 hours
                retry_filtered_models_attempts=0,
            )
        )

logger.info(dbt_cloud_logs)

Version info (prefect version output)

Version:             3.0.1
API version:         0.8.4
Python version:      3.11.9
Git commit:          c6b2ffe1
Built:               Fri, Sep 6, 2024 10:05 AM
OS/Arch:             linux/x86_64
Profile:             default
Server type:         cloud
Pydantic version:    2.8.2
Integrations:
  prefect-dbt:       0.6.0
  prefect-snowflake: 0.28.0
  prefect-shell:     0.3.0
  prefect-azure:     0.4.0

Additional context

No response

[cicdw]

Hi @bjorhn - sync/async handling was updated in 3.0 as we slowly move to a more explicit interface.

In this case, as you identified, trigger_dbt_cloud_job_run_and_wait_for_completion is an asynchronously defined flow and Prefect 3.0 removed support for calling asynchronous flow and task objects in synchronous contexts.

So you will need to await this flow for it to be executed properly.

[bjorhn]

Hey Chris - You're right. Running it in an async flow works fine, my dev just wasn't aware of some other changes related to the migration that had to be made as well (mostly related to block loading). I was mistaken about the dbt flow being sync_compatible, too. Probably shouldn't create issues when I don't have the code in front of me, got too excited about finishing the migration to 3.0

Thanks for the assistance.

[cicdw]

No worries, having it written up as an issue will probably help some other user when they search!