Task Groups

@dag(
    schedule=CronDataIntervalTimetable("0 2 * * *", timezone="UTC"),  # ds = day that closed
    start_date=datetime(2026, 6, 28, 2, tzinfo=timezone.utc),
    end_date=datetime(2026, 6, 30, 2, tzinfo=timezone.utc),  # sample orders end on 30 June
    catchup=False,
    max_active_runs=1,                            # days load in order, one dbt run at a time
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
    tags=["booknest"],
)
def booknest_daily():
    @task_group
    def ingest():
        ...                                       # extract() and load(), Section 7.5.1
        return load(extract())
    @task_group
    def model():
        dbt_build = BashOperator(
            task_id="dbt_build",
            bash_command=f"{DBT} build --select fct_sales --profiles-dir .")
        @task
        def check(ds=None) -> str:
            lines, gross = p.check_day(ds, booknest_conn())
            return f"{lines} lines, gross {gross}"
        dbt_build >> check()
    @task(outlets=[DAILY_SALES])                  # a successful publish updates the asset
    def publish(ds=None) -> int:
        return p.publish_day(ds, booknest_conn())
    ingest() >> model() >> publish()