Dagu

Appearance

Control Flow Examples ​

Examples for human tasks, conditions, repetition, routing, DAG composition, and worker placement.

Conditional Execution ​

yaml
steps:
  - run: echo "Deploying application"
    preconditions:
      - condition: "${env.ENV}"
        expected: "production"

Learn more →

Human Input Before Deployment ​

yaml
steps:
  - id: release_review
    action: human.task
    with:
      prompt: Choose the deployment environment
      form:
        type: object
        properties:
          environment:
            type: string
            enum: [staging, production]
        required: [environment]

  - id: deploy
    depends: [release_review]
    run: ./deploy.sh '${steps.release_review.outputs.environment}'
bash
dagu start --run-id release-42 release.yaml
dagu human-task complete \
  --run-id release-42 \
  --step release_review \
  --input environment=production \
  release.yaml

The first command exits after the run reaches Waiting. Completing the task enqueues the same run, and the scheduler resumes it with the validated form value available to deploy.

Learn more →

Repeat Until Condition ​

Looking for iteration over a list? See Parallel Execution.

yaml
steps:
  - run: curl -f http://service/health
    repeat_policy:
      repeat: true
      interval_sec: 10
      exit_code: [1]  # Repeat while exit code is 1

Learn more →

Repeat Until Command Succeeds ​

yaml
steps:
  - run: curl -f http://service:8080/health
    repeat_policy:
      repeat: until        # Repeat UNTIL service is healthy
      exit_code: [0]        # Exit code 0 means success
      interval_sec: 10      # Wait 10 seconds between attempts
      limit: 30            # Maximum 5 minutes

Learn more →

Repeat Until Output Match ​

yaml
 steps: 
  - run: echo "COMPLETED"  # Simulates job status check
    env:
      - JOB_STATUS: COMPLETED
    repeat_policy:
      repeat: until        # Repeat UNTIL job completes
      condition: "${env.JOB_STATUS}"
      expected: "COMPLETED"
      interval_sec: 30
      limit: 120           # Maximum 1 hour (120 attempts)

Learn more →

Repeat Steps ​

yaml
steps:
  - run: echo "heartbeat"  # Sends heartbeat signal
    repeat_policy:
      repeat: while            # Repeat indefinitely while successful
      exit_code: [0]
      interval_sec: 60

Learn more →

Repeat Steps Until Success ​

yaml
steps:
  - run: echo "Checking status"
    repeat_policy:
      repeat: until        # Repeat until exit code 0
      exit_code: [0]
      interval_sec: 30
      limit: 20            # Maximum 10 minutes

Learn more →

DAG-Level Preconditions ​

yaml
preconditions:
  - eval: "$(date +%u)"
    expected: "re:[1-5]"  # Weekdays only

steps:
  - run: echo "Run on business days"

Learn more →

Negated Preconditions ​

yaml
steps:
  # Run only when NOT in production
  - run: echo "Running dev task"
    preconditions:
      - condition: "${env.ENVIRONMENT}"
        expected: "production"
        negate: true

  # Run only on weekends
  - run: echo "Weekend maintenance"
    preconditions:
      - eval: "$(date +%u)"
        expected: "re:[1-5]"  # Weekdays
        negate: true          # Invert: run on weekends

Learn more →

Routing Based on Value ​

yaml
env:
  - STATUS: production
steps:
  - id: router
    action: router.route
    with:
      value: ${env.STATUS}
      routes:
        "production": [prod_handler]
        "staging": [staging_handler]

  - id: prod_handler
    run: echo "Production"

  - id: staging_handler
    run: echo "Staging"

Learn more →

Routing Based on Step Output ​

yaml
steps:
  - id: check_status
    run: printf 'status=success\n' >> "$DAGU_OUTPUT_FILE"
    outputs:
      - name: status

  - id: router
    action: router.route
    with:
      value: ${steps.check_status.outputs.status}
      routes:
        "success": [success_handler]
        "failure": [failure_handler]
    depends: check_status

  - id: success_handler
    run: echo "Handling success"

  - id: failure_handler
    run: echo "Handling failure"

Learn more →

Continue On: Exit Codes and Output ​

yaml
steps:
  - id: optional_check
    run: exit 3  # This will exit with code 3
    continue_on:
      exit_code: [0, 3]        # Treat 0 and 3 as non-fatal
      output:
        - "WARNING"
        - "re:^INFO:.*"       # Regex match
      mark_success: true       # Mark as success when matched
  - id: continue_after_check
    run: echo "Continue regardless"
    depends: optional_check

Learn more →

Nested Workflows ​

yaml
steps:
  - id: run_etl
    action: dag.run
    with:
      dag: etl.yaml
      params: "ENV=prod DATE=today"

  - id: run_analysis
    action: dag.run
    with:
      dag: analyze.yaml
    depends: run_etl

Learn more →

Multiple DAGs in One File ​

yaml
steps:
  - action: dag.run
    with:
      dag: data-processor
      params: "type=daily"
---
name: data-processor
params:
  - name: type
    default: batch
steps:
  - id: extract
    run: echo "Extracting ${params.type} data"

  - id: transform
    run: echo "Transforming data"
    depends: extract

Learn more →

Dispatch to Specific Workers ​

yaml
tools:
  - astral-sh/uv@0.11.14

steps:
  - id: prepare_dataset
    run: uv run --python 3.13.9 python prepare_dataset.py

  - id: train_model
    action: dag.run
    with:
      dag: train-model
    depends: prepare_dataset

  - id: evaluate_model
    action: dag.run
    with:
      dag: evaluate-model
    depends: train_model
---
name: train-model
worker_selector:
  gpu: "true"
  cuda: "11.8"
  memory: "64G"
tools:
  - astral-sh/uv@0.11.14

steps:
  - run: uv run --python 3.13.9 python train.py --gpu
---
name: evaluate-model
worker_selector:
  gpu: "true"
tools:
  - astral-sh/uv@0.11.14

steps:
  - run: uv run --python 3.13.9 python evaluate.py

Learn more →

Mixed Local and Worker Steps ​

yaml
steps:
  # Runs on any available worker (local or remote)
  - id: download_dataset
    run: wget https://data.example.com/dataset.tar.gz

  # Must run on specific worker type
  - id: process_on_gpu
    action: dag.run
    with:
      dag: process-on-gpu
    depends: download_dataset

  # Runs locally (no selector)
  - id: finish
    run: echo "Processing complete"
    depends: process_on_gpu

---
name: process-on-gpu
worker_selector:
  gpu: "true"
  gpu-model: "nvidia-a100"
tools:
  - astral-sh/uv@0.11.14

steps:
  - run: uv run --python 3.13.9 python gpu_process.py

Learn more →

Force Local Execution ​

yaml
# When default_execution_mode is "distributed", use worker_selector: local
# to keep specific DAGs on the main instance
worker_selector: local

steps:
  - id: health_check
    run: curl -f http://localhost:8080/health

  - id: finish
    run: echo "Ran locally"
    depends: health_check

Use worker_selector: local as an escape hatch in distributed deployments for lightweight DAGs that should never leave the main instance.

Learn more →

Parallel Distributed Tasks ​

yaml
tools:
  - astral-sh/uv@0.11.14

steps:
  - id: split_data
    run: |
      chunks="$(uv run --python 3.13.9 python split_data.py --chunks=10)"
      printf 'chunks=%s\n' "$chunks" >> "$DAGU_OUTPUT_FILE"
    outputs:
      - name: chunks

  - action: dag.run
    with:
      dag: chunk-processor
      params: "chunk=${ITEM}"
    parallel:
      items: ${steps.split_data.outputs.chunks}
      max_concurrent: 5
    depends: split_data

  - run: uv run --python 3.13.9 python merge_results.py
---
name: chunk-processor
worker_selector:
  memory: "16G"
  cpu-cores: "8"
params:
  - name: chunk
    default: ""
tools:
  - astral-sh/uv@0.11.14

steps:
  - run: uv run --python 3.13.9 python process_chunk.py "${params.chunk}"

Learn more →

Dagu is open source under the GNU General Public License v3.0.

Copyright © 2024 Dagu Contributors