Displaying non native XComs in a human readable way on UI

[amoghrajesh]

closes: #49456

Problem

The problem was that we used custom XCom serialisation (if defined) or regular XcomModel serialisation to serialise the xcoms before sending it across to the API server to persist to the DB.

Flow:

1. XCom is pushed via task return value or ti.xcom_push and then uses the object that is serialised using CustomXComBackend or the XCom class via the XComEncoder (e.g set, Asset, datetime) into Python JSON-serializable object.

   airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py

   Line 526 in 3208bbb

   def _xcom_push(ti: RuntimeTaskInstance, key: str, value: Any, mapped_length: int | None = None) -> None:

   airflow/task-sdk/src/airflow/sdk/bases/xcom.py

   Lines 64 to 71 in 3208bbb

   	value = cls.serialize_value(
   	value=value,
   	key=key,
   	task_id=task_id,
   	dag_id=dag_id,
   	run_id=run_id,
   	map_index=map_index,
   	)

   airflow/task-sdk/src/airflow/sdk/bases/xcom.py

   Lines 280 to 283 in 3208bbb

   	from airflow.serialization.serde import serialize
   	# return back the value for BaseXCom, custom backends will implement this
   	return serialize(value) # type: ignore[return-value]

2. FastAPI deserializes it (JsonValue) when API call is sent to API server → Converts it into a Python dict.

   airflow/task-sdk/src/airflow/sdk/api/client.py

   Line 414 in 3208bbb

   self.client.post(f"xcoms/{dag_id}/{run_id}/{task_id}/{key}", params=params, json=value)

   airflow/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py

   Lines 204 to 223 in 3208bbb

   	value: Annotated[
   	JsonValue,
   	Body(
   	description="A JSON-formatted string representing the value to set for the XCom.",
   	openapi_examples={
   	"simple_value": {
   	"summary": "Simple value",
   	"value": '"value1"',
   	},
   	"dict_value": {
   	"summary": "Dictionary value",
   	"value": '{"key2": "value2"}',
   	},
   	"list_value": {
   	"summary": "List value",
   	"value": '["value1"]',
   	},
   	},
   	),
   	],

3. SQLAlchemy stores it (Column(JSON)) → Serializes it back into JSON in the database.

   airflow/airflow-core/src/airflow/models/xcom.py

   Line 79 in 3208bbb

   value = Column(JSON().with_variant(postgresql.JSONB, "postgresql"))

But when de-serializing to show in the UI or API, we don't de-serialize using our Serialization module which happened in (1) above.

Approach

We are certain at all times that the value stored in the DB is going to be a valid JSON object, irrespective of where it got serialised from.

In airflow 3, the custom xcom backend is also only handled on the worker side and the reference for it is stored in the Xcom table in the metadata DB. So, we should not use the custom xcom backend ser / deser on the API server side at all.

A safe way to deserialize would be to pass in the full=False to the serde deserialize so that the deser can happen without loading the custom classes that COULD be present. This will return a stringified version of the object deserialized.

Since we use the "full=False", we will be returning a stringified version of the value when deserialize is set to true. A good way of handling this would be to have the UI not JSON.stringify it if it already is a string to avoid quotes display on the UI.

Testing

DAG used:

from airflow.sdk import ObjectStoragePath
from airflow import DAG
from airflow.providers.standard.operators.python import PythonOperator

def push_class(**kwargs):
    value = ObjectStoragePath("file:///tmp/foo")
    return value

with DAG(
    'xcom_push_class',
    schedule=None,
    catchup=False,
) as dag:

    push_xcom_task = PythonOperator(
        task_id='push_class',
        python_callable=push_class,
    )

    push_xcom_task

Request sent by the UI to the execution API:

curl 'http://localhost:28080/api/v2/dags/xcom_push_class/dagRuns/manual__2025-06-10T07:23:01.644733+00:00/taskInstances/push_class/xcomEntries/return_value?map_index=-1&deserialize=true&stringify=false' \
  -H 'Accept: application/json' \
  -H 'Accept-Language: en-GB,en;q=0.9' \
  -H 'Connection: keep-alive' \
  -H 'Referer: http://localhost:28080/dags/xcom_push_class/runs/manual__2025-06-10T07:23:01.644733+00:00/tasks/push_class/xcom' \
  -H 'Sec-Fetch-Dest: empty' \
  -H 'Sec-Fetch-Mode: cors' \
  -H 'Sec-Fetch-Site: same-origin' \
  -H 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36' \
  -H 'sec-ch-ua: "Google Chrome";v="137", "Chromium";v="137", "Not/A)Brand";v="24"' \
  -H 'sec-ch-ua-mobile: ?0' \
  -H 'sec-ch-ua-platform: "macOS"'

Before:

After:

TODO:

• Update PR desc
• Add testing details
• Fix tests
• Update UI code and tests to send "deserialize" flag as true

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[pierrejeambrun]

Code looks good to me, not approving just to leave time to @kaxil to check this one as he's more familiar with our xcom serialization model.

[kaxil]

some tests needs fixing, lgtm otherwise.

cc @uranusjr since he had concerns last time