Declaring a DAG & Scheduling

1DAG identity & docs#

What it does: Every DAG needs a unique identifier. You can also attach a description and tags so people (and the search box) can find it later.

Parameters:

How it works:

• Because dag_id is the primary key Maestro-Pi uses to store your DAG, if two files in the repo declare the same dag_id, whichever one is ingested last "wins" and overwrites the other in storage — keep dag_ids unique per file.
• Keep description short — it's meant for a list view, not a full README.
• owners is a common convention but is not part of Maestro-Pi's DAG-level metadata today — if you want to record an owning team, put it in description or tags as a convention.

Example:

from datetime import datetime
from dag_parser.dynamic.dag_context import DAG
from dag_parser.dynamic.operators import PythonOperator

with DAG(
    dag_id="sales_daily_report",
    description="Aggregates daily sales figures and emails a summary report",
    tags=["sales", "reporting", "daily"],
    start_date=datetime(2026, 1, 1),
) as dag:

    def build_report():
        print("building report...")

    build_report_task = PythonOperator(
        task_id="build_report",
        python_callable=build_report,
    )

2Schedule#

What it does: Decides when Maestro-Pi automatically creates a new run of your DAG. There are three ways to schedule a DAG: a cron expression, a named "timetable," or a dataset (run when upstream data changes rather than on a clock).

Parameters:

How it works:

• schedule and schedule_interval mean the same thing — just pick one name and use it consistently in a given DAG.
• Setting timetable overrides schedule entirely — a DAG uses either a cron schedule or a named timetable, never both at once.
• Passing a list of Dataset(...) objects (instead of a cron string) makes the DAG dataset-driven: it runs when the datasets it consumes get new data, not on a clock. See feature 26 for the full dataset-trigger walkthrough.
• The dataset dict form's trigger_type accepts exactly "any" (run once any listed dataset updates) or "all" (run only once every listed dataset has updated since the last successful run) — default is "all".
• Double-check your cron syntax — an invalid cron expression is only caught when the scheduler tries to evaluate it, not when the DAG is first ingested.

Example — cron:

with DAG(
    dag_id="hourly_sync",
    schedule="0 * * * *",   # every hour on the hour
    start_date=datetime(2026, 1, 1),
) as dag:
    ...

Example — cron descriptor:

with DAG(
    dag_id="daily_job",
    schedule="@daily",
    start_date=datetime(2026, 1, 1),
) as dag:
    ...

Example — named timetable:

with DAG(
    dag_id="month_end_close",
    timetable="last_day_of_month",
    start_date=datetime(2026, 1, 1),
) as dag:
    ...

Example — dataset-driven:

from dag_parser.dynamic.dag_context import Dataset

with DAG(
    dag_id="consume_sales_table",
    schedule=[Dataset("s3://bucket/sales_table")],
    start_date=datetime(2026, 1, 1),
) as dag:
    ...

# Explicit form, triggered when ANY of the listed datasets updates:
with DAG(
    dag_id="consume_any_upstream",
    schedule={
        "datasets": [Dataset("s3://bucket/table_a"), Dataset("s3://bucket/table_b")],
        "trigger_type": "any",
    },
    start_date=datetime(2026, 1, 1),
) as dag:
    ...

3Time window & timezone#

What it does: Bounds when a DAG is allowed to schedule runs, and in what timezone its cron expression should be interpreted.

Parameters:

How it works:

• start_date is the anchor Maestro-Pi uses to calculate the first (and, with catchup, every subsequent) scheduled run — without it, there's no reference point.
• timezone affects the cron's wall-clock meaning: "10 11 * * *" means 11:10 in that timezone, not UTC. If your DAG seems to run at the "wrong" hour, this is usually why.
• end_date only stops future scheduling — it does not cancel a run that's already in progress or queued.
• If you leave timezone unset, it defaults to UTC — be explicit if your team thinks in local time.

Example:

with DAG(
    dag_id="regional_batch_job",
    schedule="0 6 * * *",           # 06:00 daily...
    timezone="America/New_York",    # ...in US Eastern time, not UTC
    start_date=datetime(2026, 1, 1),
    end_date=datetime(2026, 12, 31),  # stop auto-scheduling after this date
) as dag:
    ...

4Catchup & backfill toggle#

What it does: Controls whether Maestro-Pi fills in every schedule interval that was "missed" between start_date and now, or only starts scheduling from the most recent interval going forward.

Parameters:

How it works:

• With catchup=True (the default) and a start_date set far in the past on a frequent schedule, Maestro-Pi will gradually create every missed run — a handful per scheduler cycle, not all at once — until it's caught up to "now."
• For most "just run going forward" DAGs, set catchup=False explicitly so ingesting the DAG for the first time doesn't produce a flood of historical runs.
• catchup=False still respects max_active_runs and every other run-level guardrail — it only changes where the schedule starts counting from.
• Catchup-created runs are still capped by max_active_runs (feature 5) — if the cap is already reached, the remaining catchup runs simply wait for a slot to free up.

Example:

# Backfill every missed run since start_date (default behavior)
with DAG(
    dag_id="full_history_backfill",
    schedule="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=True,
) as dag:
    ...

# Only run going forward — ignore anything missed before now
with DAG(
    dag_id="forward_only_job",
    schedule="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:
    ...