Amazon MWAA ServerlessではAirflow Variableが使えないので移行方法を考える

task = S3ListOperator(
    task_id="list_files",
    bucket="{{ var.value.data_bucket }}",
    prefix="{{ var.json.pipeline_config.prefix }}",
)

@dag(params={"bucket": "default-bucket", "prefix": "data/"})
def my_pipeline():
    task = BashOperator(
        task_id="process",
        bash_command="aws s3 ls s3://{{ params.bucket }}/{{ params.prefix }}",
    )

# テスト用ワークフロー定義
test_var:
  dag_id: test_var
  schedule: null
  default_args:
    owner: airflow
    start_date: "2024-01-01"
  tasks:
    publish_test:
      operator: airflow.providers.amazon.aws.operators.sqs.SqsPublishOperator
      task_id: publish_test
      sqs_queue: https://sqs.us-east-1.amazonaws.com/123456789012/test-queue
      message_content: '{"value":"{{ テスト対象のテンプレート変数 }}"}'

test_var_params:
  dag_id: test_var_params
  schedule: null
  default_args:
    owner: airflow
    start_date: "2024-01-01"
  params:
    greeting: "hello-from-params"
    target_bucket: "my-params-bucket"
  tasks:
    publish_params_test:
      operator: airflow.providers.amazon.aws.operators.sqs.SqsPublishOperator
      task_id: publish_params_test
      sqs_queue: https://sqs.us-east-1.amazonaws.com/123456789012/test-queue
      message_content: '{"greeting":"{{ params.greeting }}","bucket":"{{ params.target_bucket }}","ds":"{{ ds }}"}'

{
  "greeting": "hello-from-params",
  "bucket": "my-params-bucket",
  "ds": "2026-03-04"
}

test_var_value:
  dag_id: test_var_value
  schedule: null
  default_args:
    owner: airflow
    start_date: "2024-01-01"
  tasks:
    publish_var_value_test:
      operator: airflow.providers.amazon.aws.operators.sqs.SqsPublishOperator
      task_id: publish_var_value_test
      sqs_queue: https://sqs.us-east-1.amazonaws.com/123456789012/test-queue
      message_content: '{"value":"{{ var.value.test_var }}"}'

jinja2.exceptions.UndefinedError: 'var' is undefined

message_content: '{"bucket":"{{ var.json.test_json_var.bucket_name }}"}'

message_content: '{"task_id":"{{ ti.task_id }}","dag_id":"{{ ti.dag_id }}"}'

{
  "task_id": "publish_ti_test",
  "dag_id": "ex_3e5d20aa-d1fd-4c45-b636-b327eb411736",
  "ds": "2026-03-04"
}

jinja2.exceptions.UndefinedError: 'yaml_dag_handler.driver.jinja_context.SafeTaskInstance object' has no attribute 'run_id'

tasks:
  list_s3:
    operator: airflow.providers.amazon.aws.operators.s3.S3ListOperator
    task_id: list_s3
    bucket: my-bucket
    prefix: "workflows/"
  publish_result:
    operator: airflow.providers.amazon.aws.operators.sqs.SqsPublishOperator
    task_id: publish_result
    sqs_queue: https://sqs.us-east-1.amazonaws.com/123456789012/test-queue
    message_content: '{"xcom_result":"{{ ti.xcom_pull(task_ids="list_s3") }}"}'
    dependencies:
      - list_s3

{
  "xcom_result": "['workflows/sqs-sensor-basic.yaml', 'workflows/test-var-params.yaml', ...]"
}

# ds_add: 日付加減算
message_content: '{"yesterday":"{{ macros.ds_add(ds, -1) }}","tomorrow":"{{ macros.ds_add(ds, 1) }}"}'

{"yesterday": "2026-03-03", "tomorrow": "2026-03-05"}

# ds_format: フォーマット変換、random: ランダム値
message_content: >-
  {"ds_format":"{{ macros.ds_format(ds, '%Y-%m-%d', '%Y/%m/%d') }}","random":"{{ macros.random() }}"}

{"ds_format": "2026/03/04", "random": "0.7386899833460479"}

# datetime.now(), timedelta
message_content: >-
  {"datetime_now":"{{ macros.datetime.now() }}","timedelta":"{{ macros.timedelta(days=7) }}"}

{"datetime_now": "2026-03-04 12:31:34.399187", "timedelta": "7 days, 0:00:00"}

jinja2.exceptions.SecurityError: access to attribute 'strftime' of 'datetime' object is unsafe.

# DAGトップレベルでVariable.get()を呼ぶ非推奨パターン
from airflow.models import Variable

bucket = Variable.get("data_bucket")
config = Variable.get("pipeline_config", deserialize_json=True)

with DAG(dag_id="dag_with_variable", schedule="@daily", ...):
    list_files = S3ListOperator(task_id="list_files", bucket=bucket, prefix=config["prefix"])

$ dag-converter convert dag_with_variable.py

KeyError: 'Variable data_bucket does not exist'
Exception: Failed to generate dag object for file dag_with_variable.py

# Jinjaテンプレート内で var を使っているDAG
publish = SqsPublishOperator(
    task_id="publish_result",
    sqs_queue="https://sqs...",
    message_content='{"bucket":"{{ var.value.data_bucket }}"}',
)

# 変換後のYAML — {{ var.value.data_bucket }} がそのまま残る
message_content: '{"bucket":"{{ var.value.data_bucket }}"}'

my_pipeline:
  dag_id: my_pipeline
  schedule: "0 2 * * *"
  default_args:
    owner: airflow
    start_date: "2024-01-01"
  params:
    data_bucket: "default-data-bucket"
    output_prefix: "output/daily"
    max_records: "10000"
  tasks:
    list_files:
      operator: airflow.providers.amazon.aws.operators.s3.S3ListOperator
      task_id: list_files
      bucket: "{{ params.data_bucket }}"
      prefix: "{{ params.output_prefix }}"

aws mwaa-serverless start-workflow-run \
    --workflow-arn "arn:aws:airflow-serverless:us-east-1:123456789012:workflow/my-pipeline-xxx" \
    --override-parameters '{"data_bucket":"prod-data-bucket","output_prefix":"prod/output"}'

// YAML定義のparamsで実行した場合
{"greeting": "hello-from-params", "bucket": "my-params-bucket"}

// override-parametersで上書きした場合
{"greeting": "overridden-greeting", "bucket": "overridden-bucket"}

task = S3ListOperator(
    task_id="list_files",
    bucket="{{ var.value.data_bucket }}",
    prefix="{{ var.value.output_prefix }}",
)

my_pipeline:
  dag_id: my_pipeline
  params:
    data_bucket: "my-data-bucket"
    output_prefix: "output/"
  tasks:
    list_files:
      operator: airflow.providers.amazon.aws.operators.s3.S3ListOperator
      task_id: list_files
      bucket: "{{ params.data_bucket }}"
      prefix: "{{ params.output_prefix }}"

task = GlueJobOperator(
    task_id="transform",
    job_name="my-etl",
    script_args={
        "--source_bucket": "{{ var.json.pipeline_config.bucket }}",
        "--output_format": "{{ var.json.pipeline_config.format }}",
    },
)

my_pipeline:
  dag_id: my_pipeline
  params:
    source_bucket: "xxx"
    output_format: "parquet"
  tasks:
    transform:
      operator: airflow.providers.amazon.aws.operators.glue.GlueJobOperator
      task_id: transform
      job_name: my-etl
      script_args:
        "--source_bucket": "{{ params.source_bucket }}"
        "--output_format": "{{ params.output_format }}"

# Airflow UI や Secrets Manager で環境ごとに Variable を設定
# staging: data_bucket = "staging-bucket"
# production: data_bucket = "prod-bucket"
task = S3ListOperator(
    task_id="list_files",
    bucket="{{ var.value.data_bucket }}",
)

# YAML定義にはデフォルト値を設定
my_pipeline:
  params:
    data_bucket: "default-bucket"

# 環境別にパラメーターを上書きして実行
# staging
aws mwaa-serverless start-workflow-run \
    --workflow-arn "arn:aws:..." \
    --override-parameters '{"data_bucket":"staging-bucket"}'

# production
aws mwaa-serverless start-workflow-run \
    --workflow-arn "arn:aws:..." \
    --override-parameters '{"data_bucket":"prod-bucket"}'

task = SimpleHttpOperator(
    task_id="call_api",
    headers={"Authorization": "Bearer {{ var.value.external_api_key }}"},
)

# Lambda関数内でSecrets Managerから直接取得する設計に変更
my_pipeline:
  tasks:
    call_api:
      operator: airflow.providers.amazon.aws.operators.lambda_function.LambdaInvokeFunctionOperator
      task_id: call_api
      function_name: call-external-api
      payload: '{"endpoint": "/data", "method": "GET"}'

# いずれも動作しない
{{ ti.xcom_pull(task_ids='get_config').data_bucket }}      → YAMLプリプロセッサでブロック
{{ ti.xcom_pull(task_ids='get_config')['data_bucket'] }}   → 実行時エラー（XCom値が文字列のため）
{{ ti.xcom_pull(task_ids='get_config') | fromjson }}       → No filter named 'fromjson'

my_pipeline:
  tasks:
    get_bucket:
      operator: airflow.providers.amazon.aws.operators.lambda_function.LambdaInvokeFunctionOperator
      task_id: get_bucket
      function_name: get-ssm-parameter
      payload: '{"key": "/myapp/data_bucket"}'
    get_prefix:
      operator: airflow.providers.amazon.aws.operators.lambda_function.LambdaInvokeFunctionOperator
      task_id: get_prefix
      function_name: get-ssm-parameter
      payload: '{"key": "/myapp/output_prefix"}'
    list_files:
      operator: airflow.providers.amazon.aws.operators.s3.S3ListOperator
      task_id: list_files
      bucket: >-
        {{ ti.xcom_pull(task_ids='get_bucket') | replace('"', '') }}
      prefix: >-
        {{ ti.xcom_pull(task_ids='get_prefix') | replace('"', '') }}
      dependencies:
        - get_bucket
        - get_prefix

import boto3

def handler(event, context):
    ssm = boto3.client("ssm")
    resp = ssm.get_parameter(Name=event["key"])
    return resp["Parameter"]["Value"]