Airflow DbtDag supported?

[rbrtr]

Hello, i tried the airflow dbt integration and I'm not able to run the dag and the dag is not imported into airflow either. It works when I switch from DbtDag to the default DAG definition. I followed the offical documentation. https://learn.microsoft.com/en-us/fabric/data-factory/apache-airflow-jobs-dbt-fabric

This definition doesnt work

import os
from pathlib import Path
from datetime import datetime
from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig

DEFAULT_DBT_ROOT_PATH = Path(__file__).parent.parent / "dags" / "project_1"
DBT_ROOT_PATH = Path(os.getenv("DBT_ROOT_PATH", DEFAULT_DBT_ROOT_PATH))
profile_config = ProfileConfig(
     profile_name="default",
     target_name="fabric-dev",
     profiles_yml_filepath=DBT_ROOT_PATH / "profiles.yml",
)

# Define the default arguments for the DAG
dag_kd_test_project_1 = DbtDag(
     project_config=ProjectConfig(DBT_ROOT_PATH,),
     operator_args={"install_deps": True},
     profile_config=profile_config,
     schedule_interval="@daily",
     catchup=False,
     dag_id="test_project",
)

This definition works

from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator

# Define the default arguments for the DAG
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 5, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1
}

# Instantiate the DAG object
with DAG(
    'dags-helloworld.py',
    default_args=default_args,
    description='A simple Hello World DAG',
    schedule_interval="@daily",
    catchup=False
) as dag:

    # Define the tasks
    hello_task = BashOperator(
        task_id='hello_world_task',
        bash_command='echo "Hello, World!"'
    )

    # Set the task dependencies
    hello_task

[rbrtr · Accepted solution]

Adding this import solved the problem

from airflow import DAG

[DaVinci-DB]

import os
from pathlib import Path
from datetime import datetime
from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig

DEFAULT_DBT_ROOT_PATH = Path(__file__).parent.parent / "dags" / "project_1"
DBT_ROOT_PATH = Path(os.getenv("DBT_ROOT_PATH", DEFAULT_DBT_ROOT_PATH))
print(f"DBT_ROOT_PATH: {DBT_ROOT_PATH}")

profile_config = ProfileConfig(
    profile_name="default",
    target_name="fabric-dev",
    profiles_yml_filepath=DBT_ROOT_PATH / "profiles.yml",
)
print(f"Profile Config: {profile_config}")

# Define the default arguments for the DAG
dag_kd_test_project_1 = DbtDag(
    project_config=ProjectConfig(DBT_ROOT_PATH),
    operator_args={"install_deps": True},
    profile_config=profile_config,
    schedule_interval="@daily",
    catchup=False,
    dag_id="test_project",
)
print(f"DAG: {dag_kd_test_project_1}")

[Anonymous]

Hi, @rbrtr 

Is there an error message when you can't run the DAG?

For reference:

Solved: Airflow not recognizing DAGs when File storage cha... - Microsoft Fabric Community

Solved: Unauthorized Error in Airflow Task with Microsoft ... - Microsoft Fabric Community

Best Regards,

Community Support Team _Charlotte

If this post helps, then please consider Accept it as the solution to help the other members find it more quickly.