Scheduling & Triggering Runs

23Automatic runs#

What it does: How Maestro-Pi actually turns your schedule/timetable (Part 1) into real dag_run rows, with no action needed on your part.

How it works:

• schedule="@once" is special — it creates exactly one run, at start_date, the very first time the DAG is seen with no existing runs, and never again after that.
• {{ data_interval_start }} / {{ data_interval_end }} are only populated for real cron/descriptor schedules (and @once). A DAG using a named timetable= or with no schedule at all gets NULL there — avoid templating those fields on a timetable-scheduled DAG.
• max_active_runs (feature 5) is enforced at the moment a new automatic run would be created — if the DAG is already at its cap, the scheduler just tries again on the next poll rather than queuing up a backlog of "missed creations."

Example:

with DAG(
    dag_id="auto_hourly",
    schedule="0 * * * *",         # scheduler creates a run every hour, no action needed
    start_date=datetime(2026, 1, 1),
    catchup=False,
    max_active_runs=1,
) as dag:
    ...

24Manual trigger#

What it does: How a DAG gets triggered on demand (via the UI or API) with a custom conf, and how that interacts with your declared params schema (feature 8). This section covers what to know as a DAG author; the actual API call is documented separately.

How it works:

• A paused DAG cannot be manually triggered — it must be unpaused first.
• If you supply values through the typed params field, they're validated against your declared schema (required/type/enum/range). If you supply raw conf instead, it's accepted with no schema validation — useful as a workaround for the number/array/object/date/datetime type-validation gap noted in feature 8.
• Triggering is idempotent by execution date — firing the same manual trigger twice for the same resolved execution date returns the same run rather than creating a duplicate.
• A manually-triggered run is pinned to the DAG's latest version at the moment of triggering (the same version-pinning behavior mentioned in Part 0) — it won't pick up a newer version if the DAG file changes again before the run starts.

Example:

from dag_parser.dynamic.params import Param

with DAG(
    dag_id="manual_report",
    schedule=None,   # manual-trigger-only DAG
    start_date=datetime(2026, 1, 1),
    params={
        "region": Param(type="string", enum=["us", "eu", "apac"], default="us"),
        "record_limit": Param(type="integer", default=1000, minimum=1),
    },
) as dag:
    ...

# At trigger time (UI or API), supply:
#   {"params": {"region": "eu", "record_limit": 5000}}
# -> validated against the schema above, merged with any unset defaults.

25Backfill#

What it does: Creates runs for every scheduled interval in a historical date range in one request, using the DAG's own cron schedule to compute each execution date.

How it works:

• Requires a real cron/descriptor schedule — DAGs with no schedule, "@once", or a named timetable= can't be backfilled through this mechanism, even though they schedule automatic runs fine.
• There's a hard cap on how many runs a single backfill request can create — a very wide date range on a frequent schedule should be split into smaller requests.
• An optional "reset existing" mode deletes runs already in the requested range before recreating them — this is a real, irreversible delete of run history for that range, so use it deliberately.
• Backfill-created runs use their own run_type and a deterministic run ID, so re-running the same backfill request without resetting is safe — runs that already exist in the range are simply left alone.
• Backfill run creation respects the DAG's max_active_runs cap (feature 5) the same way ordinary scheduled runs do — if creating the next computed run would exceed the cap, it's held back and created once a slot frees up.

Example (conceptual — issued via API/UI, not DAG-file syntax):

# The DAG itself needs nothing special beyond a real cron schedule:
with DAG(
    dag_id="daily_aggregation",
    schedule="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,   # catchup only affects auto-scheduling, not backfill
) as dag:
    ...

26Dataset-driven trigger#

What it does: A DAG scheduled with schedule=[Dataset(...)] (feature 2) runs automatically when its upstream dataset(s) receive new data.

How it works:

• Only one dataset-triggered run is ever in flight at a time — while a run is in progress, further dataset updates don't queue up additional runs; they're coalesced into whatever the next evaluation picks up after the current run finishes.
• The "since last update" clock is anchored to the last successful run — if the most recent run failed, the same dataset events that triggered it are still considered new on the next check, so the DAG gets retried against the same updates rather than waiting for fresh ones.
• trigger_type is matched exactly against lowercase "any" or "all" — double check spelling/case if a dataset-driven DAG never seems to fire.
• A task produces a dataset event by declaring outlets=[Dataset(...)] and completing successfully — a failed producer task never emits an event, so a failure upstream naturally withholds downstream dataset-triggered DAGs.

Example:

from dag_parser.dynamic.dag_context import Dataset

# Producer: emits a dataset event on success
with DAG(dag_id="produce_sales", schedule="@daily", start_date=datetime(2026, 1, 1)) as dag:
    load = PythonOperator(
        task_id="load_sales_table",
        python_callable=lambda: None,
        outlets=[Dataset("s3://bucket/sales_table")],
    )

# Consumer: runs automatically once the dataset above gets a new event
with DAG(
    dag_id="consume_sales",
    schedule=[Dataset("s3://bucket/sales_table")],
    start_date=datetime(2026, 1, 1),
) as dag:
    ...

27Cross-DAG trigger#

What it does: TriggerDagRunOperator lets one task start a run of a different DAG, optionally waiting for it to finish before the parent task completes.

Parameters:

How it works:

• The child DAG must exist and be unpaused, or the task fails immediately with a clear error.
• conf is not validated against the child DAG's params schema the way a UI/API manual trigger is (feature 24) — a typo'd key or wrong type surfaces later, inside the child run, rather than up front.
• wait_for_completion=True does not hold a worker slot — the parent task moves into a lightweight waiting state, and a background scheduler process checks the child's state and flips the parent to success/failed later. This is safe even on a small cluster with deeply nested trigger chains.
• If the child ends up in a state that's in neither allowed_states nor failed_states, the parent simply keeps waiting — bound only by the DAG-level dagrun_timeout_seconds (feature 6) if you've set one.
• The generated child run ID follows a fixed pattern derived from the parent run — you don't choose it; read it back via XCom if a downstream task needs it.

Example:

from dag_parser.dynamic.dag_context import TriggerDagRunOperator

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

    trigger_child = TriggerDagRunOperator(
        task_id="trigger_child",
        trigger_dag_id="child_pipeline",
        conf={"batch_date": "{{ .DS }}"},   # not validated against child's params schema
        wait_for_completion=True,
        allowed_states=["success"],
        failed_states=["failed"],
    )

28Time-based defer#

What it does: Lets a task free its worker slot entirely while it waits for a condition — a timer, a specific moment, or an HTTP check — instead of holding the slot the whole time.

Parameters (self.defer(...)):

How it works:

• TimeDeltaTrigger's elapsed time is measured from the moment defer() was called, not from the run's logical date or the task's start.
• DateTimeTrigger(moment) treats a timezone-naive datetime as UTC — pass a timezone-aware datetime or an explicit UTC string if you need precision.
• HttpTrigger treats a network error as "not fired yet," not a failure — it keeps polling silently. Always set a timeout if the endpoint might be unreachable indefinitely.
• A ready-made TimeSensor (feature 49) wraps "wait until a specific time of day" without you needing to write your own defer() logic.

Example:

from dag_parser.dynamic.dag_context import BaseOperator, TimeDeltaTrigger

class DelayedStep(BaseOperator):
    operator_name = "DelayedStep"

    def execute(self, context):
        self.defer(
            trigger=TimeDeltaTrigger(300),
            method_name="execute_complete",
            timeout=600,   # give up (fail) if it hasn't fired within 10 minutes
        )

    def execute_complete(self, event=None):
        print("delay complete")