dataclasses-avroschema

Unions allow a field to hold one of several different types.

Python Version Compatibility

• Python <= 3.9: Use typing.Union[Type1, Type2].
• Python >= 3.10: Use the union type expression operator Type1 | Type2.

Union with Records

You can create unions of multiple AvroModel classes. If the models have a Meta.namespace defined, the resulting Avro schema will use the qualified name (e.g., types.Bus).

Unions with typing.Optional

typing.Optional[T] is automatically translated into a Union of T and null. In the resulting Avro schema, null is placed at the end of the type array.

# Python 3.10+ syntax for Unions
import dataclasses
from dataclasses_avroschema import AvroModel

@dataclasses.dataclass
class Bus(AvroModel):
    engine_name: str
    class Meta:
        namespace = "types"

@dataclasses.dataclass
class Car(AvroModel):
    engine_name: str
    class Meta:
        namespace = "types"

@dataclasses.dataclass
class UnionSchema(AvroModel):
    lake_trip: Bus | Car
    river_trip: Bus | Car | None = None

When using serialization_type="avro-json" with union types, the resulting JSON is not standard JSON. Each union field is wrapped in an object containing the type name (e.g., {"first_union": {"long": 10}}). This annotation is required so the deserializer knows which branch of the union to use.

Warning: Do not confuse standard JSON with avro-json. Standard JSON without these type annotations will fail to deserialize when encountering unions.

import typing
import dataclasses
import datetime
import uuid
from dataclasses_avroschema import AvroModel

@dataclasses.dataclass
class UnionSchema(AvroModel):
    first_union: typing.Union[str, int]
    logical_union: typing.Union[datetime.datetime, datetime.date, uuid.uuid4]

my_union = UnionSchema(first_union=10, logical_union=datetime.datetime.now())

# This produces annotated JSON required for deserialization
event = my_union.serialize(serialization_type="avro-json")
# Output: b'{"first_union": {"long": 10}, "logical_union": {"long": ...}}'

# Correct deserialization
my_union.deserialize(event, serialization_type="avro-json")

The serialization process is built on top of fastavro. To implement a custom serialization logic, you can override the serialize and deserialize methods in a base class that inherits from AvroModel.

Key methods to use within your override:

• self.avro_schema_to_python(): To get the schema as a Python dictionary.
• self.asdict(): To get the instance data as a dictionary.
• from_dict(...): To convert a dictionary back into a class instance during deserialization.

import dataclasses
import typing
from dataclasses_avroschema import AvroModel, SerializationType

@dataclasses.dataclass
class MyAvroModel(AvroModel):
    def serialize(self, serialization_type: SerializationType = "avro") -> bytes:
        schema = self.avro_schema_to_python()
        data = self.asdict()
        # Replace with your custom logic
        return custom_serialization(schema, data, serialization_type=serialization_type)

    @classmethod
    def deserialize(
        cls, data: bytes, serialization_type: SerializationType = "avro", create_instance: bool = True
    ) -> typing.Union[typing.Dict, "AvroModel"]:
        schema = cls.avro_schema_to_python()
        payload = custom_deserialize(data, schema, serialization_type=serialization_type)

        if create_instance:
            # Use from_dict to reconstruct the instance
            return from_dict(data_class=cls, data=payload)
        return payload

The following table describes how Avro complex types are mapped to Python types, depending on your Python version.

Python <= 3.10

Python >= 3.11