Custom registered serializers stopped working due to type changes

[FrankPortman]

Apache Airflow version

3.1.0

If "Other Airflow 2 version" selected, which one?

No response

What happened?

The way classname gets resolved has changed, breaking custom deserialize calls.

Isolated the issue to these PRs:

• https://github.com/apache/airflow/pull/52360/files#diff-5d7a43b9860e3c6d80d8feadce5ae76a3a56698fd8dff389fb6095c01153e74eR267
• Enable Serde for Pydantic BaseModel and Subclasses #51059

Which mentions pydantic only but has a classname --> cls change for custom deserializers bundled in the commentary. The pydantic specific naming of the PRs made this difficult to debug when looking at git history.

The documentation also still shows comparing classname to qualname(...), as well as str type hints for classname. This also made it difficult to debug when poring over changes for 3.1.0.

What you think should happen instead?

Release notes and documentation (including this example, which is basically the only example of registering serializers: https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/serializers.html) should be updated if behavior changes, and this should have been listed as a significant/breaking change in 3.1.0 release notes.

How to reproduce

Use registered custom serializers and follow the official documentation for the deserialize function.

Operating System

Any

Versions of Apache Airflow Providers

No response

Deployment

Astronomer

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[vikramkoka]

cc: @amoghrajesh @kaxil

[kaxil]

cc @sjyangkevin @bolkedebruin

[sjyangkevin]

Can I work on the issue and will also update the document, sorry for introducing the breaking change.

[FrankPortman]

Thanks for the quick response!

[sjyangkevin]

Hi @FrankPortman

I just had a deeper look into the codebase and the document. I think the change to function signature for the registered deserializers might not break the custom deserialize call. Meanwhile, I am trying to set up an example and test it. Below is my finding and feel free to correct me (or I might correct myself after running a concrete example). It would also be helpful if you had a breaking custom deserializer example (or a sample DAG) and I can take this to troubleshoot as well.

1. The function signature of the deserialize method

airflow/airflow-core/src/airflow/serialization/serde.py

Line 196 in a6506f2

def deserialize(o: T | None, full=True, type_hint: Any = None) -> object:

2. The registered deserializer, where the cls is passed instead of classname.

airflow/airflow-core/src/airflow/serialization/serde.py

Lines 265 to 267 in 9d4447d

	# registered deserializer
	if classname in _deserializers:
	return _deserializers[classname].deserialize(cls, version, deserialize(value))

3. If the data being passed has a custom deserializer, and does not match to any registered deserializer, the following will be used

airflow/airflow-core/src/airflow/serialization/serde.py

Lines 272 to 274 in 9d4447d

	# class has deserialization function
	if hasattr(cls, "deserialize"):
	return getattr(cls, "deserialize")(deserialize(value), version)

If a custom deserialization is needed and if the class that the data belongs to doesn't match to any registered deserializers. The case 3 should be invoked. In this case, getattr(cls, "deserialize") gets the deserialize method from the class, taking the official document, the deserialize method from the Foo class (https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/serializers.html#airflow-object).

https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/serializers.html#registered for me looks more like an example shows how the serializer/deserializer is registered in airflow, but not the instruction for defining a custom serializer/deserializer in user code. However, this need to be updated to reflect the latest changes.

I will follow up back to this issue once I get my environment built and able to run tests.

[sjyangkevin]

I just done some testing by following the document: https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/serializers.html#serialization. Below is my step-by-step walk through my testing process.

First, I define a class (i.e., my data required custom serialization) without defining the serialize and deserialize methods. The expectation is to fail both serialization and deserialization since no registered serializer/deserializer is defined for it.

class Foo:
    __version__: ClassVar[int] = 1

    def __init__(self, a, v) -> None:
        self.a = a
        self.b = {"x": v}

It shows the failure at serialization as expected.

Then, I added the serialize and the deserialize method, and I would expect the deserialize to fail, because this class is not part of the allowed_deserialization_classes.

from typing import Any, ClassVar


class Foo:
    __version__: ClassVar[int] = 1

    def __init__(self, a, v) -> None:
        self.a = a
        self.b = {"x": v}

    def serialize(self) -> dict[str, Any]:
        return {
            "a": self.a,
            "b": self.b,
        }

    @staticmethod
    def deserialize(data: dict[str, Any], version: int):
        f = Foo(a=data["a"])
        f.b = data["b"]
        return f

The deserialization failed as expected.

Then, I added the class to the allowed_deserialization_classes through environment variable.

export AIRFLOW__CORE__ALLOWED_DESERIALIZATION_CLASSES_REGEXP=".*\.Foo"

The task failed unexpectedly with the following error:

If we look closer to the deserialize definition in the class Foo.

@staticmethod
    def deserialize(data: dict[str, Any], version: int):
        # `v` is a mandatory parameter in the class init method, but it is not passed during initialization
        f = Foo(a=data["a"])
        f.b = data["b"]
        return f

Then, if we make v an optional parameter, or pass a value for v, the serialization/deserialization will work as expected. I feel like it is inaccurate documentation instead of code issue. Feel free to let me know any feedback. I would like to help further to resolve any potential issue.

Below is my testing DAG code.

from airflow.decorators import dag, task
from pendulum import datetime

from typing import Any, ClassVar


class Foo:
    __version__: ClassVar[int] = 1

    def __init__(self, a, v = None) -> None:
        self.a = a
        self.b = {"x": v}

    def serialize(self) -> dict[str, Any]:
        return {
            "a": self.a,
            "b": self.b,
        }

    @staticmethod
    def deserialize(data: dict[str, Any], version: int):
        f = Foo(a=data["a"])
        f.b = data["b"]
        return f



@dag(
    start_date=datetime(2025, 5, 23),
    schedule=None,
    catchup=False,
    tags=["serialization", "pydantic", "airflow"],
)
def pydantic_serde():
    
    ... # other task definitions

    # 12. Test custom serializer/deserializer
    @task
    def get_custom_class():
        foo = Foo(a="1", v="2")
        print("custom class:", foo)
        return foo
    
    @task
    def print_custom_class(foo):
        print("get the custom class:", foo)

    # DAG chaining
    ... # other test cases
    print_custom_class(get_custom_class())

pydantic_serde()

[FrankPortman]

@sjyangkevin maybe my original comment was ambiguous but by custom registered serializers I mean the ones that are defined in a separate file and placed in the Airflow namespace, not the ones that are methods on the objects themselves.

[sjyangkevin]

@FrankPortman , thanks for the clarification. I am just trying to understand a bit more, probably there are some usage patterns that I am not aware of. Do you mean the registered serializers defined here: https://github.com/apache/airflow/tree/main/airflow-core/src/airflow/serialization/serializers?

Or, do you refer to the custom serializers you defined, and then registered similarly as registering a plugin or listener?