"""
DAG для запуска dbt run команд
"""
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'dbt_run_example',
    default_args=default_args,
    description='Пример DAG для запуска dbt run',
    schedule_interval=None,  # Запуск вручную
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['dbt', 'example'],
) as dag:

    # Проверка версии dbt
    dbt_debug = BashOperator(
        task_id='dbt_debug',
        bash_command='cd /dbt && dbt debug --project-dir . --profiles-dir /home/airflow/.dbt',
    )

    # Установка зависимостей dbt
    dbt_deps = BashOperator(
        task_id='dbt_deps',
        bash_command='cd /dbt && dbt deps --project-dir . --profiles-dir /home/airflow/.dbt',
    )

    # Запуск dbt run
    dbt_run = BashOperator(
        task_id='dbt_run',
        bash_command='cd /dbt && dbt run --project-dir . --profiles-dir /home/airflow/.dbt',
    )

    # Запуск dbt test (опционально)
    dbt_test = BashOperator(
        task_id='dbt_test',
        bash_command='cd /dbt && dbt test --project-dir . --profiles-dir /home/airflow/.dbt',
    )

    # Определяем порядок выполнения задач
    dbt_debug >> dbt_deps >> dbt_run >> dbt_test