amoghrajesh commented on code in PR #46719:
URL: https://github.com/apache/airflow/pull/46719#discussion_r1957178671
##########
task_sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -346,11 +348,15 @@ def get_relevant_upstream_map_indexes(
def _xcom_push(ti: RuntimeTaskInstance, key: str, value: Any, mapped_length:
int | None = None) -> None:
# Private function, as we don't want to expose the ability to manually set
`mapped_length` to SDK
# consumers
- from airflow.models.xcom import XCom
+ from airflow.serialization.serde import serialize
# TODO: Move XCom serialization & deserialization to Task SDK
# https://github.com/apache/airflow/issues/45231
- value = XCom.serialize_value(value)
+
+ # The execution API server now deals in json compliant objects.
+ # It is responsibility of the client to handle any non native object
serialization.
+ # serialize does just that.
+ value = serialize(value)
Review Comment:
For the custom xcom backend, we anyways store ONLY the `serialised` path in
the DB right?
Even with this change, we will store the same but not in a serialised
manner, so those interactions will now happen via python strings. That should
still be OK i guess
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
