dependencies between assets in airflow

[MammadTavakoli]

I define two assets in airflow 3.0.3 that after running first asset (user) second asset (user_location) run too. the code is like this:

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)

    return data

@asset(
    schedule=user,    
)
def user_location(self, context: Context, user: Asset) -> dict[str]:       
    user_data = context['ti'].xcom_pull(
        dag_id="user",
        task_ids="user",           
        include_prior_dates=True
    )
    return user_data['results'][0]['location']

but after running the code following error happened:

[2025-08-03, 08:36:15] INFO - DAG bundles loaded: dags-folder, example_dags: source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-08-03, 08:36:15] INFO - Filling up the DagBag from /opt/airflow/dags/user.py: source="airflow.models.dagbag.DagBag"
[2025-08-03, 08:36:24] INFO - Task instance is in running state: chan="stdout": source="task"
[2025-08-03, 08:36:24] INFO -  Previous state of the Task instance: TaskInstanceState.QUEUED: chan="stdout": source="task"
[2025-08-03, 08:36:24] INFO - Current task name:user_location: chan="stdout": source="task"
[2025-08-03, 08:36:24] INFO - Dag name:user_location: chan="stdout": source="task"
[2025-08-03, 08:36:24] ERROR - Task failed with exception: source="task"
TypeError: 'NoneType' object is not subscriptable
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 877 in run

File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1164 in _execute_task

File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 397 in wrapper

File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/standard/operators/python.py", line 217 in execute

File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/standard/operators/python.py", line 240 in execute_callable

File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/callback_runner.py", line 81 in run

File "/opt/airflow/dags/user.py", line 35 in user_location

[2025-08-03, 08:36:24] INFO - Task instance in failure state: chan="stdout": source="task"
[2025-08-03, 08:36:24] INFO - Task start: chan="stdout": source="task"
[2025-08-03, 08:36:24] INFO - Task:<Task(_AssetMainOperator): user_location>: chan="stdout": source="task"
[2025-08-03, 08:36:24] INFO - Failure caused by 'NoneType' object is not subscriptable: chan="stdout": source="task"

How can solve the problem?

[gopidesupavan]

There is likely issue with include_prior_dates that was fixed lastweek, #53809 it should be available in next release.
Can you try without include_prior_dates is it working? on 3.0.3

BTW i tested your example on main branch its working fine. not with S3 similar i mimicked

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)

    return {"name": "John Doe", "age": 30, "location": "New York"}


@asset(
    schedule=user,
)
def user_location(self, context: Context, user: Asset) -> dict[str]:
    user_data = context['ti'].xcom_pull(
        dag_id="user",
        task_ids="user",
        include_prior_dates=True
    )
    print(f"User data pulled from XCom: {user_data}")
    return user_data

[MammadTavakoli]

I tried your code without include_prior_dates, but the result is same.
also i installed airflow in docker

[gopidesupavan]

not sure i am not able to reproduce, its working fine for me. i would request to test on the latest rc release #54124