# dags/orders_aggregate.py
# Hourly aggregation of orders.events into warehouse.daily_orders.
# Owned by the data-platform team. Last edited: 2026-04-12.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
from etl.aggregate import build_daily_orders

default_args = {
    "owner": "data-platform",
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    dag_id="orders_aggregate",
    default_args=default_args,
    start_date=datetime(2026, 4, 1),
    schedule_interval="@hourly",
    catchup=False,
    max_active_runs=1,
) as dag:

    def _run(execution_date, **_):
        start = execution_date - timedelta(hours=1)
        end = execution_date
        rows = build_daily_orders(start=start, end=end)
        print(f"wrote {rows} rows for window {start} -> {end}")

    aggregate = PythonOperator(
        task_id="aggregate",
        python_callable=_run,
    )