Part 2

Controlling a DAG Run's Behavior

5Run concurrency limits#

What it does: Two independent caps: how many runs of this DAG can be active at once, and how many tasks (across all its active runs, combined) can be running at once.

Parameters:

Param	Type	Default	Notes
`max_active_runs`	integer	`16`	Cap on concurrent `dag_run`s for this DAG.
`max_active_tasks`	integer	`None` (no DAG-level cap)	Cap on concurrent task instances across all of this DAG's active runs combined.

How it works:

If max_active_runs is reached, Maestro-Pi simply doesn't create the next scheduled run yet — it waits for a slot to free up rather than skipping or queuing indefinitely.
max_active_tasks applies across all runs of the DAG combined, not per-run — with several concurrent runs and a low cap, tasks from different runs compete for the same budget.
Set these thoughtfully on a wide, fan-out-heavy DAG — a low max_active_tasks throttles a single run's own internal parallelism, not just cross-run parallelism.

Example:

python

with DAG(
    dag_id="throttled_ingest",
    schedule="*/15 * * * *",
    start_date=datetime(2026, 1, 1),
    max_active_runs=1,     # never more than 1 run of this DAG in flight at once
    max_active_tasks=5,    # never more than 5 tasks (across all its runs) running at once
) as dag:
    ...

6Run timeout#

What it does: Force-fails an entire DAG run if it's still going after a set number of seconds — a hard ceiling on total run duration.

Parameters:

Param	Type	Default	Notes
`dagrun_timeout_seconds`	integer (seconds)	`None`	If unset, falls back to a global 24h abandoned-run safety net (see below).

How it works:

This is a strict wall-clock timeout — it fires even if every task inside the run is healthy and actively progressing. Set it generously if your DAG legitimately runs long.
If you don't set it, the run still isn't unbounded — there's a global 24-hour fallback, but that one only fires if the run has genuinely gone quiet (no task actively heartbeating) — a slow-but-healthy run is never killed by the fallback alone.
On timeout, the run and any still-active child tasks are force-failed — there's no "soft warning" state, it's a hard stop.
This is separate from a task's own execution_timeout (feature 16) — set both if you want protection at both the run level and the individual task level.

Example:

python

with DAG(
    dag_id="nightly_etl",
    schedule="@daily",
    start_date=datetime(2026, 1, 1),
    dagrun_timeout_seconds=3600 * 4,   # force-fail the whole run if it exceeds 4 hours
) as dag:
    ...

7DAG-level SLA#

What it does: Flags (but never fails) a run that takes longer than expected — a monitoring signal, not an enforcement mechanism.

Parameters:

Param	Type	Default	Notes
`expected_duration_seconds`	integer (seconds)	`None`	Threshold above which the run is flagged as an SLA miss.
`on_sla_miss_callback`	`SmtpNotifier(...)` (or a dict via the task-level `_callbacks` pattern — see feature 21)	`None`	What to do when the SLA is missed.

How it works:

An SLA miss is detection-only — it never changes the run's state or stops it. If you want to actually kill a long-running run, use dagrun_timeout_seconds (feature 6) instead.
It fires once per run, not repeatedly on every subsequent scheduler tick.
This measures the whole run's elapsed time; there's a separate, independent task-level SLA (feature 17) for flagging individual slow tasks.
For reliable delivery today, use SmtpNotifier(...) (or the task-level dict form) for your callback — see feature 9 for the full picture on DAG-level callbacks.

Example:

python

from dag_parser.dynamic.dag_context import SmtpNotifier

with DAG(
    dag_id="revenue_pipeline",
    schedule="@hourly",
    start_date=datetime(2026, 1, 1),
    expected_duration_seconds=600,   # flag as an SLA miss if a run takes > 10 minutes
    on_sla_miss_callback=SmtpNotifier(
        to=["team@company.com"],
        subject="SLA miss: revenue_pipeline",
        html_content="<p>This run exceeded its expected duration.</p>",
    ),
) as dag:
    ...

8Typed run parameters#

What it does: Declares the shape of the conf a user must supply when manually triggering the DAG — types, defaults, enums, ranges, and which fields are required. The UI builds a form straight from this schema.

Parameters (fields of Param(...)):

Field	Type	Default	Notes
`type`	string	required	One of `string`, `integer`, `number`, `boolean`, `array`, `object`, `date` (`YYYY-MM-DD`), `datetime` (ISO string). Any other value breaks ingestion.
`default`	matches `type`	none	If set, the param is treated as optional.
`required`	boolean	inferred	No `default` → required. Any `default` (even `0`/`False`/`""`) → optional, unless you set `required=False`/`True` explicitly.
`enum`	list	none	Restricts allowed values.
`minimum` / `maximum`	number	none	For `integer`/`number`.
`min_length` / `max_length`	integer	none	For `string`.
`pattern`	regex string	none	For `string`.
`description`	string	`""`	Shown in the UI form.

How it works:

enum/range/pattern rules are checked when someone actually triggers a run, not when the DAG file is parsed — a bad conf is rejected at trigger time.
Booleans are checked strictly — True/False will not be silently accepted where an integer/number is expected, and vice versa.
Inside your tasks, you read the values supplied at trigger time through templating ({{ .Params.key }} in Go-templated fields, or params.key in Jinja) — the Param(...) object itself only defines the schema, not the runtime value.
All eight declared types (string, integer, number, boolean, array, object, date, datetime) are fully validated against your schema whenever a value is supplied through the typed params field at trigger time.

Example:

python

from dag_parser.dynamic.params import Param

with DAG(
    dag_id="etl_orders",
    schedule=None,             # manually triggered only
    start_date=datetime(2026, 1, 1),
    params={
        "run_date": Param(type="string", required=True, description="Business date, YYYY-MM-DD",
                           pattern=r"^\d{4}-\d{2}-\d{2}$"),
        "customer_id": Param(type="integer", default=0, minimum=0),
        "mode": Param(type="string", enum=["full", "incremental"], default="incremental"),
        "full_load": Param(type="boolean", default=False),
        "threshold": Param(type="number", default=0.95, minimum=0.0, maximum=1.0),
    },
) as dag:
    ...

9DAG-level callbacks#

What it does: Attaches an alert to a DAG run reaching a terminal outcome — on_success_callback, on_failure_callback, and on_sla_miss_callback (feature 7).

Parameters:

Param	Type	Default	Notes
`on_success_callback`	callable, or `SmtpNotifier(...)`	`None`	Fires when a run finishes successfully.
`on_failure_callback`	callable, or `SmtpNotifier(...)`	`None`	Fires when a run ends in failure.
`on_sla_miss_callback`	callable, or `SmtpNotifier(...)`	`None`	Fires on an SLA miss (feature 7).

How it works:

Pass either a plain Python function (it receives a context dict describing the run) for custom logic, or a SmtpNotifier(to=[...], subject=..., html_content=...) for a ready-made email alert — both dispatch correctly.
If you need a DAG-level outcome to trigger a Slack, HTTP webhook, or PagerDuty alert declaratively rather than writing your own function, the task-level callback pattern in feature 21 supports those channels directly — or add a dedicated notification task at the end of the DAG (e.g. SlackAPIPostOperator with trigger_rule="all_done" or "one_failed").
For per-task alerting (a specific task's own success/failure/retry/skip), use the task-level callback pattern in feature 21 instead.

Example:

python

from dag_parser.dynamic.dag_context import SmtpNotifier

def notify_failure(context):
    print(f"DAG failed: {context['run_id']}")

with DAG(
    dag_id="critical_pipeline",
    schedule="@daily",
    start_date=datetime(2026, 1, 1),
    on_success_callback=SmtpNotifier(
        to=["team@company.com"],
        subject="critical_pipeline succeeded",
        html_content="<p>Run completed successfully.</p>",
    ),
    on_failure_callback=notify_failure,
) as dag:
    ...

10Access control in DAG file#

What it does: Declares per-role permissions for this specific DAG directly in code, layered on top of (and separate from) permissions set manually in the Admin UI.

Parameters:

Param	Type	Default	Notes
`access_control`	dict `{role_name: [permissions]}`	`{}`	Permissions: `can_read`, `can_trigger`, `can_edit`, `can_delete`, `can_clear`.

How it works:

role_name must match an existing role — one of the built-in Admin, Op, Editor, Viewer, Public, or a custom role created via the Admin UI.
Re-ingesting the DAG file replaces the set of permissions it manages each time — if you remove access_control from the file, those grants are removed on the next ingestion pass.
Permissions set manually through the Admin UI are never touched by this — the two layers coexist per role/DAG.
This only scopes access to this DAG — broader, cross-DAG role permissions are a separate, global RBAC concern.

Example:

python

with DAG(
    dag_id="finance_close",
    schedule="@monthly",
    start_date=datetime(2026, 1, 1),
    access_control={
        "Editor": ["can_read", "can_trigger", "can_edit"],
        "Viewer": ["can_read"],
        "Op": ["can_read", "can_trigger", "can_clear"],
    },
) as dag:
    ...

11Partitioning#

What it does: Scopes a DAG's runs to a partition key (a date bucket, a region, a customer segment, etc.), so each run's data footprint is tracked independently of plain run history.

Parameters:

Param	Type	Default	Notes
`partitions`	one of `DailyPartition()`, `HourlyPartition()`, `WeeklyPartition()`, `MonthlyPartition()`, `StaticPartition([...])`	`None`	See below for how the key is computed for each.

How it works:

For the time-based types (Daily/Hourly/Weekly/Monthly), the partition key is derived automatically from the run's execution date — e.g. DailyPartition() produces "2026-07-09".
StaticPartition([...]) is different: it does not derive a key from the execution date at all — you must supply partition_key explicitly when triggering a run for a statically-partitioned DAG. Pass a non-empty list of keys.
Partitioning is orthogonal to scheduling — you still need a schedule/ timetable/dataset trigger (or manual triggers) to actually create runs; partitions only changes how those runs are tagged and tracked.

Example:

python

from dag_parser.dynamic.dag_context import DailyPartition, StaticPartition

# Time-based: partition_key auto-derived from execution_date, e.g. "2026-07-09"
with DAG(
    dag_id="daily_regional_load",
    schedule="@daily",
    start_date=datetime(2026, 1, 1),
    partitions=DailyPartition(),
) as dag:
    ...

# Static: partition_key must be supplied explicitly at trigger time
with DAG(
    dag_id="per_region_backfill",
    schedule=None,
    start_date=datetime(2026, 1, 1),
    partitions=StaticPartition(["us-east", "us-west", "eu-central"]),
) as dag:
    ...

12Jinja customization#

What it does: Four DAG-level knobs that change how Python-family operators render their fields through Jinja2 at execution time.

Parameters:

Param	Type	Default	Notes
`render_template_as_native_obj`	boolean	`False`	If `True`, a template resolving to a single value returns a real Python object (`int`, `list`, ...) instead of always a string.
`user_defined_macros`	dict `{name: callable}`	`{}`	Extra names usable inside `{{ }}` expressions.
`user_defined_filters`	dict `{name: callable}`	`{}`	Extra `\|filter` functions.
`template_undefined`	a Jinja2 `Undefined` subclass (e.g. `jinja2.StrictUndefined`)	permissive (renders empty)	Pass the class itself, not an instance.

How it works:

These four settings apply only to Python-family operators (PythonOperator/ExternalPythonOperator/PythonVirtualenvOperator) — specifically their op_args/op_kwargs/templates_dict. Non-Python operators (Bash/SQL/HTTP/etc.) render through a separate Go-templating mechanism (feature 39) that doesn't read these settings at all.
render_template_as_native_obj=True only changes behavior for a template that's just one expression (e.g. "{{ params.count }}"). A mixed template like "count={{ params.count }}" always renders as a string, flag or not.
Because these attributes are read by re-importing the DAG file at execution time, your DAG file must remain importable on the worker at that moment — keep it valid even after you've edited other parts of it.

Example:

python

import jinja2

def to_upper(value):
    return str(value).upper()

with DAG(
    dag_id="native_templated_job",
    schedule="@daily",
    start_date=datetime(2026, 1, 1),
    render_template_as_native_obj=True,
    user_defined_macros={"env_name": "production"},
    user_defined_filters={"upper": to_upper},
    template_undefined=jinja2.StrictUndefined,   # fail loudly on unknown template vars
) as dag:

    def process(count, label):
        print(f"{label}: {count} (type={type(count)})")

    PythonOperator(
        task_id="process",
        python_callable=process,
        op_kwargs={
            "count": "{{ params.count }}",              # renders as a real int (native obj)
            "label": "{{ env_name | upper }}",           # macro + custom filter
        },
    )