apiserver INFO: 2401:4900:883b:3304:a978:c82c:e446:6ff:0 - "GET /d7xzx8ch/api/v2/dags/attach_extra_info/dagRuns/manual__2025-09-17T11%3A26%3A10.871455%2B00%3A00/taskInstances/attach_extra_using_metadata/xcomEntries/pipeline_outlets?map_index=-1&deserialize=true&stringify=false HTTP/1.1" 500 Internal Server Error
apiserver ERROR: Exception in ASGI application
apiserver + Exception Group Traceback (most recent call last):
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/_utils.py", line 79, in collapse_excgroups
apiserver | yield
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/middleware/base.py", line 183, in __call__
apiserver | async with anyio.create_task_group() as task_group:
apiserver | ^^^^^^^^^^^^^^^^^^^^^^^^^
apiserver | File "/usr/local/lib/python3.12/site-packages/anyio/_backends/_asyncio.py", line 772, in __aexit__
apiserver | raise BaseExceptionGroup(
apiserver | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
apiserver +-+---------------- 1 ----------------
apiserver | Traceback (most recent call last):
apiserver | File "/usr/local/lib/python3.12/site-packages/uvicorn/protocols/http/httptools_impl.py", line 409, in run_asgi
apiserver | result = await app( # type: ignore[func-returns-value]
apiserver | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apiserver | File "/usr/local/lib/python3.12/site-packages/uvicorn/middleware/proxy_headers.py", line 60, in __call__
apiserver | return await self.app(scope, receive, send)
apiserver | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apiserver | File "/usr/local/lib/python3.12/site-packages/fastapi/applications.py", line 1054, in __call__
apiserver | await super().__call__(scope, receive, send)
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/applications.py", line 113, in __call__
apiserver | await self.middleware_stack(scope, receive, send)
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/middleware/errors.py", line 186, in __call__
apiserver | raise exc
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/middleware/errors.py", line 164, in __call__
apiserver | await self.app(scope, receive, _send)
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/middleware/gzip.py", line 29, in __call__
apiserver | await responder(scope, receive, send)
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/middleware/gzip.py", line 130, in __call__
apiserver | await super().__call__(scope, receive, send)
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/middleware/gzip.py", line 46, in __call__
apiserver | await self.app(scope, receive, self.send_with_compression)
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/middleware/cors.py", line 85, in __call__
apiserver | await self.app(scope, receive, send)
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/middleware/base.py", line 182, in __call__
apiserver | with recv_stream, send_stream, collapse_excgroups():
apiserver | ^^^^^^^^^^^^^^^^^^^^
apiserver | File "/usr/local/lib/python3.12/contextlib.py", line 158, in __exit__
apiserver | self.gen.throw(value)
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/_utils.py", line 85, in collapse_excgroups
apiserver | raise exc
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/middleware/base.py", line 184, in __call__
apiserver | response = await self.dispatch_func(request, call_next)
apiserver | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apiserver | File "/usr/local/lib/python3.12/site-packages/astronomer/runtime/plugin.py", line 90, in dispatch
apiserver | response = await call_next(request)
apiserver | ^^^^^^^^^^^^^^^^^^^^^^^^
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/middleware/base.py", line 159, in call_next
apiserver | raise app_exc
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/middleware/base.py", line 144, in coro
apiserver | await self.app(scope, receive_or_disconnect, send_no_error)
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/middleware/exceptions.py", line 63, in __call__
apiserver | await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
apiserver | raise exc
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/routing.py", line 78, in app
apiserver | await wrap_app_handling_exceptions(app, request)(scope, receive, send)
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
apiserver | raise exc
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
apiserver | await app(scope, receive, sender)
apiserver | File "/usr/local/lib/python3.12/site-packages/starlette/routing.py", line 75, in app
apiserver | response = await f(request)
apiserver | ^^^^^^^^^^^^^^^^
apiserver | File "/usr/local/lib/python3.12/site-packages/fastapi/routing.py", line 328, in app
apiserver | content = await serialize_response(
apiserver | ^^^^^^^^^^^^^^^^^^^^^^^^^
apiserver | File "/usr/local/lib/python3.12/site-packages/fastapi/routing.py", line 182, in serialize_response
apiserver | return field.serialize(
apiserver | ^^^^^^^^^^^^^^^^
apiserver | File "/usr/local/lib/python3.12/site-packages/fastapi/_compat.py", line 152, in serialize
apiserver | return self._type_adapter.dump_python(
apiserver | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apiserver | File "/usr/local/lib/python3.12/site-packages/pydantic/type_adapter.py", line 572, in dump_python
apiserver | return self.serializer.to_python(
apiserver | ^^^^^^^^^^^^^^^^^^^^^^^^^^
apiserver | pydantic_core._pydantic_core.PydanticSerializationError: Unable to serialize unknown type: <class 'airflow.sdk.definitions.asset.Asset'>
"""
### Attach additional information to Airflow Datasets
This DAG attaches extra information to an Airflow Dataset from within the
producing task using two different methods.
"""
from airflow.decorators import dag, task
from airflow.datasets import Dataset
from pendulum import datetime
# import the Metadata class
from airflow.datasets.metadata import Metadata
my_dataset_1 = Dataset("x-dataset-metadata-1")
my_dataset_2 = Dataset("x-dataset-metadata-2")
@dag(
start_date=datetime(2024, 8, 1),
schedule=None,
catchup=False,
tags=["datasets"],
default_args={"retries": 2},
)
def attach_extra_info():
@task(outlets=[my_dataset_1])
def attach_extra_using_metadata():
num = 23
yield Metadata(my_dataset_1, {"myNum": num})
return "hello :)"
attach_extra_using_metadata()
@task(outlets=[my_dataset_2])
def use_outlet_events(**context):
num = 42
context["outlet_events"][my_dataset_2].extra = {
"myNum": num,
"myStr": "Lemons!",
}
return "hello :)"
use_outlet_events()
attach_extra_info()
Apache Airflow version
3.1.0b2
If "Other Airflow 2 version" selected, which one?
No response
What happened?
What you think should happen instead?
No response
How to reproduce
Use below dag and perform the steps:
Operating System
Linux
Versions of Apache Airflow Providers
No response
Deployment
Other
Deployment details
No response
Anything else?
No response
Are you willing to submit PR?
Code of Conduct