updated comms to support subclasses of generic types

[astro-anand]

Updated _msgpack_enc_hook to support sending subclasses of builtin types

closes: #53474

---

[astro-anand]

@ashb let me know if this is what you had in mind

[RNHTTR]

I wonder if the user should be responsible for casting to a supported built-in. In the case of the issue (i.e. using np.float64), what if they expect and want to use float64 downstream when accessing the extra? I guess that's not a big deal since extra here should just be metadata.

If we proceed with this, I think we need to update docs to highlight that types that can be forced into a builtin will be forced into a builtin

[amoghrajesh]

It actually seems that the asset metadata never reaches BaseSerialization.serialize() - it fails much earlier during pydantic's default JSON serialization when sending the HTTP API request. We probably should send it through the serialization layer similar to xcoms.

[astro-anand]

@RNHTTR @amoghrajesh You're both totally right. I think we should ensure that the serializer is called and ensure that on the other side of things, the deserialize method is called correctly too. Will update and test today

[astro-anand]

I've tried updating the _serialize_outlet_events method to use the serialize function defined in serde.py, but I'm still getting the same error.

def _serialize_outlet_events(events: OutletEventAccessorsProtocol) -> Iterator[dict[str, Any]]:
    from airflow.serialization.serde import serialize

    if TYPE_CHECKING:
        assert isinstance(events, OutletEventAccessors)
    # We just collect everything the user recorded in the accessors.
    # Further filtering will be done in the API server.
    for key, accessor in events._dict.items():
        if isinstance(key, AssetUniqueKey):
            yield {"dest_asset_key": serialize(key), "extra": serialize(accessor.extra)}
        for alias_event in accessor.asset_alias_events:
            yield attrs.asdict(alias_event)

[amoghrajesh]

You have found a good bug.

import numpy as np
npf = np.float64(1.23412415)
isinstance(npf, _primitives)
Out[5]: True
npi = np.int64(290)
isinstance(npi, _primitives)
Out[7]: False

It never reaches the np serializer. Feel free to create an issue for that.

[astro-anand]

Issue created and fixed here. Once merged, will update the outlet events with serialization & deserialization. Food for thought, do you think that asset metadata should respect the configured Xcom backend? Or is it fine that this will all be persisted in the metadata db?

[uranusjr]

We are not allowed to use non-sdk modules in sdk code. Why does this need to use deserialize (and serialize below)?

[astro-anand]

Yes you're right. This PR is not ready yet. I was trying to test if we could pass the Asset metadata through the serialization layer (like XComs) per @amoghrajesh's recommendation and this implementation was just meant to be a quick check if it would work. What do you think the best way to reuse existing serializers would be?

[astro-anand]

@uranusjr - I was roughly trying to replicate the xcom serialize_value functionality here. Won't this also be problematic since non-sdk code can't be used in the task sdk?

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

Line 336 in f77baa2

from airflow.serialization.serde import serialize

[uranusjr]

There was a decision some time ago to move serde to SDK so it should be fine to use it in execution_time now. cc @amoghrajesh to confirm.

[amoghrajesh]

@astro-anand serde has now been moved into task sdk: https://github.com/apache/airflow/tree/main/task-sdk/src/airflow/sdk/serde. You can rebase and import from there.

[github-actions]

This pull request has been closed because the author has not responded to a request for more information.