GitHub user MammadTavakoli created a discussion: dependencies between assets in
airflow
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
# Define your MinIO bucket and key (path within the bucket)
MINIO_BUCKET_NAME = "csv"
MINIO_KEY = "users.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?
GitHub link: https://github.com/apache/airflow/discussions/54064
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
