airflow_dbt_cloud_job_trigger_operator_with_status_monitoring.py

1import os
2from datetime import datetime
3
4from airflow.models import DAG
5from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator
6
7# Requirements: 
8# 1. A dbt Cloud connection (default: dbt_cloud_default) with an API token.
9# 2. The job_id for the dbt Cloud job you wish to trigger.
10
11DBT_CLOUD_ACCOUNT_ID = os.getenv("DBT_CLOUD_ACCOUNT_ID")
12DBT_CLOUD_JOB_ID = os.getenv("DBT_CLOUD_JOB_ID")
13
14with DAG(
15    dag_id="example_dbt_cloud_operator",
16    start_date=datetime(2021, 1, 1),
17    schedule_interval=None,
18    catchup=False,
19) as dag:
20
21    # Trigger a dbt Cloud job run
22    trigger_job_run = DbtCloudRunJobOperator(
23        task_id="trigger_job_run",
24        account_id=DBT_CLOUD_ACCOUNT_ID,
25        job_id=DBT_CLOUD_JOB_ID,
26        check_interval=60,
27        timeout=3600,
28    )
29
30    trigger_job_run