"""Airflow-shaped DAG definition. A PARSING fixture: it is never imported by a scheduler here,
and every operator is the no-op EmptyOperator, so the file describes a topology and does nothing.

Topology (shared with the serialized-JSON, Graphviz, Mermaid and Argo fixtures in this category):

    ingest_orders -> validate_orders -> transform_orders -+-> load_warehouse    -+-> notify_owner
                                                          +-> refresh_dashboard -+
"""

from __future__ import annotations

import datetime as dt

from airflow import DAG
from airflow.operators.empty import EmptyOperator

DEFAULT_ARGS = {
    "owner": "example-data-team",
    "retries": 2,
    "retry_delay": dt.timedelta(minutes=5),
    "depends_on_past": False,
}

with DAG(
    dag_id="orders_etl",
    description="Nightly orders extract, validate, transform and publish.",
    schedule="17 2 * * *",
    start_date=dt.datetime(2026, 1, 1),
    catchup=False,
    max_active_runs=1,
    default_args=DEFAULT_ARGS,
    tags=["orders", "etl", "example"],
) as dag:
    ingest_orders = EmptyOperator(task_id="ingest_orders")
    validate_orders = EmptyOperator(task_id="validate_orders")
    transform_orders = EmptyOperator(task_id="transform_orders")
    load_warehouse = EmptyOperator(task_id="load_warehouse")
    refresh_dashboard = EmptyOperator(task_id="refresh_dashboard")
    notify_owner = EmptyOperator(task_id="notify_owner", trigger_rule="all_done")

    ingest_orders >> validate_orders >> transform_orders
    transform_orders >> [load_warehouse, refresh_dashboard] >> notify_owner