prefect_flow_trigger_dbt_run_with_dbtcoreoperation_block.py

1from prefect import flow
2from prefect_dbt.cli.commands import DbtCoreOperation
3
4@flow
5def trigger_dbt_run():
6    # Define the dbt command to run
7    result = DbtCoreOperation(
8        commands=["dbt run"],
9        project_dir="path/to/your/dbt/project",
10        profiles_dir="path/to/your/dbt/profiles"
11    ).run()
12    
13    return result
14
15if __name__ == "__main__":
16    trigger_dbt_run()