dbt cloud sensor py file

dbt_cloud_sensor.py

from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator
from airflow.hooks.base_hook import BaseHook
from config.current import DBT_CLOUD_CONNECTION_ID


class DBTCloudSensor:
    def __init__(self):
        self.dbt_cloud_connection = BaseHook.get_connection(DBT_CLOUD_CONNECTION_ID)
        self.account_id = self.dbt_cloud_connection.extra_dejson.get('account_id')

    def dbt_cloud_run_job_operator(self, task_id, job_id, dag):
        """
        Creates and returns a DbtCloudRunJobOperator task.
        """
        return DbtCloudRunJobOperator(
            task_id=task_id,
            job_id=job_id,
            account_id=self.account_id,
            dbt_cloud_conn_id=DBT_CLOUD_CONNECTION_ID,
            check_interval=60,
            wait_for_termination=True,
            dag=dag,
        )