# dags/us_rates_elt_dag.py

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta
import os

# Путь к твоим скриптам внутри контейнера Airflow
# здесь лежат extract.py, transform.py, load.py
SCRIPTS_PATH = "/opt/airflow/etl_process"

default_args = {
    'owner': 'Ilias',
    'depends_on_past': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=1)
}

with DAG(
    'us_rates_elt_full',
    default_args=default_args,
    description='Full ELT DAG for US rates CSV',
    start_date=datetime(2026, 1, 24, 0, 0),
    schedule_interval='*/5 * * * *',  # каждые 5 минут
    catchup=False,
    max_active_runs=1
) as dag:

    # --- Extract task ---
    t1 = BashOperator(
        task_id='extract',
        bash_command=f'python {os.path.join(SCRIPTS_PATH, "extract.py")}'
    )

    # --- Transform task ---
    t2 = BashOperator(
        task_id='transform',
        bash_command=f'python {os.path.join(SCRIPTS_PATH, "transform.py")}'
    )

    # --- Load task ---
    t3 = BashOperator(
        task_id='load',
        bash_command=f'python {os.path.join(SCRIPTS_PATH, "load.py")}'
    )

    # Определяем последовательность задач
    t1 >> t2 >> t3