Airflow UI触发DAG时如何将配置列表值转为List类型？

核心问题原因 #

通过Airflow UI传入的列表配置，使用{{ dag_run.conf['input'] }}模板渲染后会被转为字符串类型，无法直接遍历，需要转换回原生List类型。

解决方法 #

方法1：直接从上下文获取原类型（推荐） #

在PythonOperator的回调函数中，直接通过context获取dag_run.conf，无需模板渲染，能直接拿到原生的List类型：

from airflow.decorators import dag, task
from datetime import datetime

@dag(start_date=datetime(2023, 1, 1), schedule=None)
def my_dag():
    @task
    def process_input(**context):
        # 直接获取原类型的列表，无需转换
        script_input = context['dag_run'].conf.get('input', [])
        for item in script_input:
            print(f"Date: {item['date']}, Update: {item['update']}, Delete: {item['delete']}")

    process_input()

my_dag()

方法2：使用json.loads解析字符串 #

如果必须通过模板渲染获取值，可结合tojson过滤器和json模块解析：

from airflow.decorators import dag, task
from datetime import datetime
import json

@dag(start_date=datetime(2023, 1, 1), schedule=None)
def my_dag():
    @task
    def process_input(**kwargs):
        # 模板渲染后得到JSON格式字符串
        script_input_str = kwargs['templates_dict']['input_str']
        # 解析为List类型
        script_input = json.loads(script_input_str)
        for item in script_input:
            # 处理逻辑
            pass

    # 传递模板变量时使用tojson过滤器
    process_input(templates_dict={'input_str': "{{ dag_run.conf['input'] | tojson }}"})

my_dag()

方法3：使用ast.literal_eval解析 #

若字符串是Python字面量格式，可用ast模块安全解析：

from airflow.decorators import dag, task
from datetime import datetime
import ast

@dag(start_date=datetime(2023, 1, 1), schedule=None)
def my_dag():
    @task
    def process_input(**kwargs):
        script_input_str = kwargs['templates_dict']['input_str']
        # 解析为List类型
        script_input = ast.literal_eval(script_input_str)
        for item in script_input:
            # 处理逻辑
            pass

    process_input(templates_dict={'input_str': "{{ dag_run.conf['input'] }}"})

my_dag()

注意事项 #

• 确保传入的配置JSON格式合法，无语法错误，否则解析会抛出异常
• 优先使用方法1，避免字符串转换带来的潜在问题，代码更简洁可靠

内容的提问来源于stack exchange，提问作者shah