managed-airflow-dag-authoring

SKILL.md

GCP Managed Airflow DAG Authoring Guide

This skill guides you through authoring and validating Apache Airflow DAGs for
Managed Service for Apache Airflow (MSAA; formerly Cloud Composer) environments.

Phase 1: Context Discovery

Before writing any DAG code, you MUST understand the constraints (e.g. version
of Airflow) and capabilities of your target environment if user is willing to
provide them.

1.1 Identify Target Environment & Access

Determine if you have direct access to the target Managed Airflow environment,
local development environment or if you are working offline (only changing local
files without validation).

If environment access is available: Use gcloud to inspect the
environment (see Section 1.3).

If offline: Rely on user provided details.

1.2 Identify Development Environment

Determine if a local development environment is available.

Check if composer-dev CLI is installed.

Check if a local Python environment with airflow is available.

1.3 Inspect Target Environment (if available and requested)

Run the following commands to discover version constraints:

Get Airflow/Image Version:

gcloud composer environments describe {env_name} \
    --location {region} \
    --format="value(config.softwareConfig.imageVersion)"

Get Installed Packages (Versions):

gcloud composer environments describe {env_name} \
    --location {region} \
    --format="value(config.softwareConfig.pypiPackages)"

Get DAGs GCS Bucket:

gcloud composer environments describe {env_name} \
    --location {region} \
    --format="value(config.dagGcsPrefix)"

Phase 2: DAG Authoring Best Practices

2.1 General Airflow Best Practices

Idempotency: Every task SHOULD be idempotent. Running it multiple times
with the same inputs (e.g., execution date) SHOULD produce the same result
and not duplicate data.

No Top-Level Code Execution: Do NOT execute database queries, external
API calls, or heavy computations at the top level of the DAG file (outside
of tasks/operators). This code runs every few seconds during DAG parsing and
will degrade performance.

Explicit Catchup: Always set catchup=False in the DAG definition
unless historical backfilling is explicitly required.

Use Airflow Variables/Connections: Never hardcode credentials or
environment-specific configs. Use Variable.get() (with
deserialize_json=True if applicable) and BaseHook.get_connection().
Access variables via Jinja templates (e.g., {{ var.value.my_var }}) to
avoid database calls during DAG parsing.

2.2 Airflow 2 vs Airflow 3 Compatibility

Use managed-airflow-migrations skill to navigate adjusting the code to
specific target Airflow version.

Phase 3: Validation Process

You MUST validate DAGs before concluding your task.

3.1 Local Validation (Offline/Pre-deployment)

3.1.1 Static Analysis & Linting

Use ruff or pylint if available.

ruff check path/to/dag.py

If targeting Airflow 3, check with Airflow 3 rules if rulesets are
available.

3.1.2 Local Dev Environment (composer-dev)

If the user has composer-dev configured:

Copy the DAG to the local directory with DAGs:

cp path/to/dag.py $(composer-dev describe {local_env} --format="value(dags_directory)")

Verify parsing:

composer-dev run-airflow-cmd {local_env} dags list-import-errors

3.2: Target Environment Validation

Only perform these steps if you have GCP access and are authorized to deploy to
a target environment.

3.2.1 Deploy to GCS

Upload the DAG to the target environment's GCS bucket:

gcloud storage cp path/to/dag.py gs://{target_bucket}/dags/

3.2.2 Verify via Airflow CLI

Wait 1-2 minutes for the scheduler to parse the file, then run:

Check for Import Errors:

gcloud composer environments run {env_name} \
    --location {region} \
    dags list-import-errors

Pass Criteria: Output should be "No data found" or empty.

Verify DAG is Listed:

gcloud composer environments run {env_name} \
    --location {region} \
    dags list | grep {dag_id}

3.2.3 Monitor Cloud Logging

Check for runtime parsing errors in Cloud Logging:

resource.type="cloud_composer_environment"
resource.labels.environment_name="{env_name}"
log_id("airflow-scheduler")
severity>=ERROR
textPayload:"{dag_file_name}"

Definition of Done

DAG code adheres to Airflow version constraints of the target environment.

DAG code follows best practices (no top-level execution, idempotent if
possible).

DAG parses locally without import errors.

(If environment is available) DAG is deployed to the target environment and
verified to have no import errors.