Airflow 2.3.3中For循环遇NoneType不可迭代错误求助

问题根源 #

你在使用update_teradata.expand(consultas=consulta_afip())时，Airflow的expand会把consulta_afip()返回的列表拆分成多个独立的update_teradata任务实例，每个实例只接收列表中的单个元素，而非整个列表。这导致你在update_teradata里执行df = pd.DataFrame(consultas)时，输入的consultas是单个JSON对象（而非列表），生成的DataFrame结构和本地测试时完全不同，最终遍历df_consulta时出现NoneType报错。

修复方案 #

根据你的需求，有两种处理方式：

方式1：取消并行，处理完整列表 #

如果不需要并行处理每个API返回结果，直接去掉expand，让update_teradata接收完整的结果列表：

# 替换原expand调用
update_teradata(consultas=consulta_afip())

方式2：并行处理单个元素 #

如果要保留并行执行，修改update_teradata来处理单个Contribuyente对象，而非整个列表：

@task
def update_teradata(consulta):
    # 直接处理单个返回对象，无需转成DataFrame
    datos = {
        'idpersona': '',
        'tipopersona': '',
        'estadoclave': '',
        'nombre': '',
        'nombreprovincia': '',
        'localidad': '',
        'codpostal': '',
        'direccion': ''
    }

    # 提取顶层字段
    for n in consulta:
        if n.lower() in datos:
            datos[n.lower()] = consulta.get(n)
    # 提取嵌套的domicilioFiscal字段
    n_fiscal = consulta['domicilioFiscal']
    for f in n_fiscal:
        if f.lower() in datos:
            datos[f.lower()] = n_fiscal.get(f)
    
    # 转换为DataFrame行（若需返回给后续任务）
    d_f = pd.DataFrame([datos])
    columnas_teradata = ['CUIT', 'SITUACION_JURIDICA', 'ESTADO', 'NOMBRE_COMPLETO', 'PROVINCIA', 'LOCALIDAD',
                         'CODIGO_POSTAL', 'DIRECCION']
    d_f.columns = columnas_teradata

    return d_f

# 保留expand，每个任务处理单个元素
update_teradata.expand(consultas=consulta_afip())

额外注意事项 #

• 禁止在Airflow任务中使用global变量（如global tsql），任务是独立执行的，全局变量会引发不可预期的问题，建议将游标定义在函数内部。
• 本地测试时要模拟Airflow的expand行为，比如传入单个元素测试update_teradata，而非整个列表，这样能提前发现结构不匹配的问题。

内容的提问来源于stack exchange，提问作者Ayelén Paris