---
name: trigger-dlt-airflow-dag
description: Trigger and monitor Airflow DAGs in a local Docker Compose setup via the scheduler container CLI. Use when running a DAG manually, checking DAG run status, monitoring task logs, or debugging a failed DAG run with docker exec.
disable-model-invocation: true
argument-hint: "[dag-id]"
---

# Trigger Airflow DAG

Trigger an Airflow DAG running in a local Docker Compose setup and optionally monitor it until completion.

## Context

When Airflow runs in Docker Compose, the CLI is accessible through the scheduler container. This skill uses `docker exec` to interact with the Airflow CLI without needing API credentials or direct access to the webserver.

## Steps

1. Find the scheduler container name.

   ```bash
   docker ps --format "table {{.Names}}\t{{.Image}}" | grep -i scheduler
   ```

2. If `$ARGUMENTS` is not provided, list all available DAGs and ask which one to trigger.

   ```bash
   docker exec <scheduler_container> airflow dags list 2>/dev/null
   ```

3. Trigger the DAG.

   ```bash
   docker exec <scheduler_container> airflow dags trigger <dag_id>
   ```

   To pass configuration to the DAG run:

   ```bash
   docker exec <scheduler_container> airflow dags trigger <dag_id> --conf '{"key": "value"}'
   ```

4. Check the DAG run status.

   ```bash
   # Preferred: JSON output (Airflow 2.6+) — reliable parsing
   docker exec <scheduler_container> airflow dags list-runs --dag-id <dag_id> --output json 2>/dev/null

   # Fallback: table output — look for the state column
   docker exec <scheduler_container> airflow dags list-runs --dag-id <dag_id> 2>/dev/null | head -5
   ```

   States: `queued` → `running` → `success` or `failed`.

5. To wait for the DAG to complete, poll until the status changes.

   ```bash
   DAG_ID="<dag_id>"
   CONTAINER="<scheduler_container>"
   RUN_ID="<run_id_from_trigger>"
   while true; do
     STATUS=$(docker exec $CONTAINER airflow dags list-runs --dag-id $DAG_ID --output json 2>/dev/null | python3 -c "import sys,json; runs=[r for r in json.load(sys.stdin) if r.get('run_id','')=='$RUN_ID']; print(runs[0]['state'] if runs else 'unknown')" 2>/dev/null)
     # Fallback if --output json is not available:
     # STATUS=$(docker exec $CONTAINER airflow dags list-runs --dag-id $DAG_ID 2>/dev/null | grep "$RUN_ID" | awk '{print $5}')
     echo "$(date): $DAG_ID status: $STATUS"
     if [ "$STATUS" = "success" ] || [ "$STATUS" = "failed" ]; then
       echo "DAG run finished with status: $STATUS"
       break
     fi
     sleep 15
   done
   ```

6. If the DAG failed, inspect task logs to find the failing task.

   ```bash
   # List task instances for the run
   docker exec <scheduler_container> airflow tasks states-for-dag-run <dag_id> <execution_date> 2>/dev/null

   # Read logs for a specific task
   docker exec <scheduler_container> airflow tasks log <dag_id> <task_id> <execution_date> 2>/dev/null
   ```

## Validation

- [ ] Scheduler container found
- [ ] DAG triggered successfully
- [ ] DAG run completed (if monitoring was requested)

## Examples

**Other useful DAG commands:**

```bash
# Pause a DAG
docker exec <scheduler_container> airflow dags pause <dag_id>

# Unpause a DAG
docker exec <scheduler_container> airflow dags unpause <dag_id>

# Show DAG details
docker exec <scheduler_container> airflow dags show <dag_id>
```

**Notes:**

- The scheduler container is preferred over the webserver for CLI commands
- Deprecation warnings in the output can be ignored
- DAG runs are created with `externally triggered: True`