Deserialize response of get_all when we call XCom.get_all

[amoghrajesh]

closes: #53018

The data returned from the api comes in without deserialisation as the contract between task sdk and the api server is simple: json objects.

So the returned data must be deserialised.

DAG used:

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

from datetime import datetime, timedelta

twos_day = datetime.now() - timedelta(days=2)

def native_obj_false(x, y):
    return x, y


def check_type(**context):
    ti = context["ti"]
    pulled_value = ti.xcom_pull(
        task_ids="py_native_obj_false", key="return_value"
    )
    print(f"This is the datatype for the pulled xcom value: {type(pulled_value)}")
    print(f"This is datatype for the 0th index of the xcom value: {type(pulled_value[0])}")
    print(f"This is datatype for the 1st index of the xcom value: {type(pulled_value[0])}")
    assert type(pulled_value[0]) == str and type(pulled_value[1]) == str


default_args = {"owner": "airflow", "depends_on_past": True}

with DAG(
    dag_id="native_obj_false",
    start_date=twos_day,
    schedule=None,
    # when set to false only the __str__() method is returned
    render_template_as_native_obj=False,
    tags=["core", "negative"],
    user_defined_macros={
        "says_hi": "Hello World!",
        "the_planets": [
            "Mercury",
            "Venus",
            "Earth",
            "Mars",
            "Jupiter",
            "Saturn",
            "Neptune",
        ],
        "ints": [1, 2, 3, 4, 5],
    },
) as dag:

    t1 = PythonOperator(
        task_id="py_native_obj_false",
        python_callable=native_obj_false,
        op_args=["{{ says_hi }}", "{{ ints }}"],
    )

    t2 = PythonOperator(
        task_id="check_xcoms",
        python_callable=check_type,
    )

t1 >> t2

Before:

After changes, dag run:

---

^ 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.

[eladkal]

Should we have a test for it?

[amoghrajesh]

Should we have a test for it?

yep, was just finishing it off.

[vatsrahul1001]

Verified re-running failed Dags works fine now

[kaxil]

The code change looks good to me. Test works but can be improved later on

https://github.com/apache/airflow/pull/53020/files#r2194788728

[amoghrajesh]

@kaxil please merge this if all's good :)

[github-actions]

Backport successfully created: v3-0-test

Status	Branch	Result
✅	v3-0-test