Airflow 3.0.3资产依赖执行报错：'NoneType'不可下标如何解决？

问题现象 #

在Airflow 3.0.3中定义了两个asset：user和user_location，期望user执行完成后自动触发user_location运行，但执行user_location时出现以下错误：

TypeError: 'NoneType' object is not subscriptable

错误指向user_data['results'][0]['location']代码行，说明user_data为None。

错误原因 #

1. 不符合Airflow 3.x资产设计逻辑：手动通过xcom_pull获取上游资产结果是冗余操作，Airflow资产依赖支持自动参数注入，不需要手动操作XCom
2. DAG ID指定错误：每个asset对应的DAG名称并非直接等于函数名，手动指定dag_id="user"会导致拉取不到正确的XCom数据
3. 参数使用不当：include_prior_dates=True可能拉取到历史未完成或无关的任务实例数据，进一步导致返回结果为None

解决方案 #

直接利用Airflow 3.x的资产依赖注入特性，让上游user资产的返回值自动传递给user_location，无需手动处理XCom。修正后的代码如下：

from airflow.sdk import asset, Asset, Context
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
import json
    
@asset(
        uri="s3://csv/users.json", 
        schedule="@daily",
        group='test'
    )
def user(self):
    s3_path = self.uri.replace("s3://", "")
    bucket, key = s3_path.split("/", 1)

    hook = S3Hook(aws_conn_id='minio_conn')

    content = hook.read_key(key=key, bucket_name=bucket)

    data = json.loads(content)

    # 提前校验数据结构，避免下游访问出错
    if "results" not in data or not data["results"]:
        raise ValueError("用户数据格式错误，缺少有效results字段")
    return data

@asset(
    schedule=[user],  # 列表格式兼容多依赖场景，单个依赖也可直接写user
)
def user_location(self, user: dict) -> dict:       
    # 直接使用注入的user参数获取上游结果
    return user['results'][0]['location']

关键修改说明 #

• 移除user_location中的context参数和xcom_pull代码，通过参数注入直接获取上游user资产的返回值
• 将schedule=user调整为schedule=[user]，符合Airflow 3.x资产依赖的推荐格式
• 在user函数中增加数据结构校验，提前发现数据问题，避免下游无意义报错
• 调整user_location的参数类型标注为dict，贴合实际传递的数据类型

额外验证步骤 #

1. 确认user资产执行成功，在Airflow UI的资产页面可查看其最新状态为已完成
2. 检查user返回的数据结构，确保存在results[0]['location']的合法访问路径
3. 查看user_location的依赖关系，确认其已正确关联user资产

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