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

BASE_DIR = "/opt/airflow"

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

with DAG(
    dag_id='employee_etl',
    default_args=default_args,
    start_date=datetime(2026, 1, 1),
    schedule_interval='*/1 * * * *',
    catchup=False,
    max_active_runs=1
) as dag:

    extract_task = BashOperator(
        task_id='extract',
        bash_command=f'python {os.path.join(BASE_DIR, "process/extract.py")}'
    )

    transform_task = BashOperator(
        task_id='transform',
        bash_command=f'python {os.path.join(BASE_DIR, "process/transform.py")}'
    )

    load_task = BashOperator(
        task_id='load',
        bash_command=f'python {os.path.join(BASE_DIR, "process/load.py")}'
    )

    extract_task >> transform_task >> load_task