composer-dag-builderlisted
Install: claude install-skill rk-chavali/gcp-de-skills
# Composer DAG builder
Read `references/conventions.md` first. DAG id pattern is
`<domain>_<cadence>_<purpose>`.
## Default DAG skeleton
```python
from datetime import datetime, timedelta
from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import (
BigQueryInsertJobOperator,
)
from airflow.providers.google.cloud.operators.cloud_run import (
CloudRunExecuteJobOperator,
)
from airflow.providers.google.cloud.sensors.bigquery import (
BigQueryTablePartitionExistenceSensor,
)
PROJECT_ID = "acme-shop-prod"
REGION = "us-central1"
default_args = {
"owner": "data-platform",
"retries": 3,
"retry_delay": timedelta(minutes=5),
"retry_exponential_backoff": True,
"max_retry_delay": timedelta(minutes=30),
"execution_timeout": timedelta(hours=1),
"depends_on_past": False,
"email_on_failure": False,
}
with DAG(
dag_id="orders_daily_load",
description="Loads storefront orders into mart_retail.fct_order.",
start_date=datetime(2026, 1, 1),
schedule="0 6 * * *",
catchup=False,
max_active_runs=1,
default_args=default_args,
tags=["retail", "daily", "owner:data-platform"],
sla_miss_callback=None,
) as dag:
wait_for_raw = BigQueryTablePartitionExistenceSensor(
task_id="wait_for_raw_orders",
project_id=PROJECT_ID,
dataset_id="raw_storefront",
table_id="orders",
partition_id="{{ ds_nodash }}",
mode="reschedule",
poke_interval=300,