prefect_dbt_core_operation_debug_and_run_flow.py

1from prefect import flow
2from prefect_dbt.cli.commands import DbtCoreOperation
3
4@flow
5def trigger_dbt_flow():
6    # Run dbt debug to check connection
7    result = DbtCoreOperation(
8        commands=["dbt debug", "dbt run"],
9        project_dir="path/to/my/dbt_project",
10        profiles_dir="path/to/my/dbt_project"
11    ).run()
12    
13    return result
14
15if __name__ == "__main__":
16    trigger_dbt_flow()