import logging
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.utils.context import Context
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)

def create_incident(context: Context):
    try:
        dag_id = context["task_instance"].dag_id
        task_id = context["task_instance"].task_id
        log_url = context["task_instance"].log_url
        logger.info(f"creating an incident for task: {task_id} of DAG: {dag_id} and url is {log_url}")
    except Exception as e:
        logger.error(f"Failed to send failure email: {e}")

def fail_task():
    raise Exception("This task is designed to fail.")

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2025, 6, 24, 0, 0),
    'retries': 0,
    'retry_delay': timedelta(minutes=5),
    'on_failure_callback': create_incident,
}

with DAG(
    'fail_dag',
    default_args=default_args,
    description='A simple DAG that fails all the time',
    schedule_interval=timedelta(days=1),
) as dag:

    fail_task = PythonOperator(
        task_id='always_fail_task',
        python_callable=fail_task
    )