Repository: marcosschroh/dataclasses-avroschema

Issue:
Avro schemas containing a union where two subschemas have same fields but different name result in incorrect serialization
**Context**
We use this library in combination with Kafka where we have a topic containing multiple type of events. All must be on the same topic since we need the ordering.

**Describe the bug**
Union types where subschemas have the same fields and types do not produce the correct binary output according to the avro specification. This results in losing compatability with other languages following the specification. The problem is being caused by the following code in `AvroBaseModel`:

```py
def serialize(self, serialization_type: str = AVRO) -> bytes:
    """
    Overrides the base AvroModel's serialize method to inject this
    class's standardization factory method
    """
    schema = self.avro_schema_to_python()

    return serialization.serialize(
        self.asdict(standardize_factory=self.standardize_type),
        schema,
        serialization_type=serialization_type,
    )
```

By transforming the pydantic object to a dict and passing it further, the typing information is lost. The fastavro library will determine the index by finding the type in the union that has the most fields in common. As the example below demonstrates, all fields are the same and it will just pick the first one.

A possible solution would be to iterate over the schema after turning it to a dict and then add the type information as the first item in the tuple. Fastavro will be able to handle that and will use the type information to determine the int for the binary output instead of the most fields. See the write_union in fastavro below:

```py
def write_union(encoder, datum, schema, named_schemas, fname, options):
    """A union is encoded by first writing a long value indicating the
    zero-based position within the union of the schema of its value. The value
    is then encoded per the indicated schema within the union."""

    best_match_index = -1
    if isinstance(datum, tuple) and not options.get("disable_tuple_notation"):
        (name, datum) = datum
        for index, candidate in enumerate(schema):
            extracted_type = extract_record_type(candidate)
            if extracted_type in NAMED_TYPES:
                schema_name = candidate["name"]
            else:
                schema_name = extracted_type
            if name == schema_name:
                best_match_index = index
                break

        if best_match_index == -1:
            field = f"on field {fname}" if fname else ""
            msg = (
                f"provided union type name {name} not found in schema "
                + f"{schema} {field}"
            )
            raise ValueError(msg)
        index = best_match_index
    else:
        pytype = type(datum)
        most_fields = -1
```

**To Reproduce**
Consider the following schema (very minimal example):

```avsc
{
  "namespace": "com.example",
  "type": "record",
  "name": "TopicEvents",
  "fields": [
    {
      "name": "event", "type": [
        "EventOne",
        "EventTwo",
      ]
    }
  ]
}

{
  "namespace": "com.example",
  "type": "record",
  "name": "EventOne",
  "fields": [{ "name": "field1", "type": "string", "schema_tag": "type": "string" }]
}

{
  "namespace": "com.example",
  "type": "record",
  "name": "EventTwo",
  "fields": [{ "name": "field1", "type": "string", "schema_tag": "type": "string" }]
}
```

An example that is incorrect would be:

```py
from typing import Literal
from pydantic import Field
from dataclasses_avroschema.pydantic.main import AvroBaseModel

class EventOne(AvroBaseModel):
    field1: str = Field(...)
    schema_tag: Literal["EventOne"] = Field(
        default="EventOne"
    )

class EventTwo(AvroBaseModel):
    field1: str = Field(...)
    schema_tag: Literal["EventTwo"] = Field(
        default="EventTwo"
    )

class Events(AvroBaseModel):
    event: EventOne | EventTwo = Field(discriminator="schema_tag")
    
serialized = Events(event=EventTwo(field1="hello world")).serialize()

print(serialized)
```

The result is: `b'\x00\x16hello world\x10EventTwo'`, but the expected result would be: `b'\x02\x16hello world\x10EventTwo'` according to the avro specification, since the int in the beginning indicates the index of the type of the union. Currently, the result of serializing `Events(event=EventOne(field1="hello world")).serialize()` and `Events(event=EventTwo(field1="hello world")).serialize()` is the same.

**Expected behavior**
Events(event=EventTwo(field1="hello world")).serialize() should result in the following value:`b'\x02\x16hello world\x10EventTwo'`

**As soon as we figured out a solution and agree on how to fix it, i can help with the implementation if necessary**

Discussion:
github-actions[bot]: [PR Preview Action](https://github.com/rossjrw/pr-preview-action) v1.4.8
:---:
:rocket: Deployed preview to https://marcosschroh.github.io/dataclasses-avroschema/pr-preview/pr-751/
on branch [`gh-pages`](https://github.com/marcosschroh/dataclasses-avroschema/tree/gh-pages) at 2024-09-20 16:22 UTC
<!-- Sticky Pull Request Commentpr-preview -->
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/marcosschroh/dataclasses-avroschema/pull/751?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Marcos+Schroh) Report
Attention: Patch coverage is `95.65217%` with `1 line` in your changes missing coverage. Please review.
> Project coverage is 99.44%. Comparing base [(`879fe7e`)](https://app.codecov.io/gh/marcosschroh/dataclasses-avroschema/commit/879fe7e4c2b105982de0861e9291f1d526b24ee6?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Marcos+Schroh) to head [(`0e98416`)](https://app.codecov.io/gh/marcosschroh/dataclasses-avroschema/commit/0e98416f7c1c678b33063c34a9c55e123df94985?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Marcos+Schroh).

| [Files with missing lines](https://app.codecov.io/gh/marcosschroh/dataclasses-avroschema/pull/751?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Marcos+Schroh) | Patch % | Lines |
|---|---|---|
| [dataclasses\_avroschema/utils.py](https://app.codecov.io/gh/marcosschroh/dataclasses-avroschema/pull/751?src=pr&el=tree&filepath=dataclasses_avroschema%2Futils.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Marcos+Schroh#diff-ZGF0YWNsYXNzZXNfYXZyb3NjaGVtYS91dGlscy5weQ==) | 91.66% | [1 Missing :warning: ](https://app.codecov.io/gh/marcosschroh/dataclasses-avroschema/pull/751?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Marcos+Schroh) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master     #751      +/-   ##
==========================================
- Coverage   99.49%   99.44%   -0.06%     
==========================================
  Files          34       34              
  Lines        1970     1966       -4     
==========================================
- Hits         1960     1955       -5     
- Misses         10       11       +1     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/marcosschroh/dataclasses-avroschema/pull/751?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Marcos+Schroh).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Marcos+Schroh).


Patch:
diff --git a/dataclasses_avroschema/__init__.py b/dataclasses_avroschema/__init__.py
index 80659a8..12aa797 100644
--- a/dataclasses_avroschema/__init__.py
+++ b/dataclasses_avroschema/__init__.py
@@ -1,12 +1,5 @@
 from .schema_generator import AvroModel  # noqa: I001
-from .types import (
-    Int32,
-    Float32,
-    TimeMicro,
-    DateTimeMicro,
-    condecimal,
-    confixed
-)
+from .types import Int32, Float32, TimeMicro, DateTimeMicro, condecimal, confixed
 from .model_generator.generator import BaseClassEnum, ModelType, ModelGenerator
 from .fields.field_utils import (
     BOOLEAN,
@@ -132,4 +125,4 @@ __all__ = [
     "DecimalField",
     "RecordField",
     "AvroField",
-]
\ No newline at end of file
+]
diff --git a/dataclasses_avroschema/faust/main.py b/dataclasses_avroschema/faust/main.py
index ce0d948..c1d0600 100644
--- a/dataclasses_avroschema/faust/main.py
+++ b/dataclasses_avroschema/faust/main.py
@@ -26,13 +26,18 @@ class AvroRecord(Record, AvroModel):  # type: ignore
         schema = self.avro_schema_to_python()
         return validate(self.asdict(), schema)
 
-    def standardize_type(self) -> typing.Any:
+    def standardize_type(self, include_type: bool = True) -> typing.Any:
         """
         Standardization factory that converts data according to the
         user-defined pydantic json_encoders prior to passing values
         to the standard type conversion factory
         """
-        return standardize_custom_type(self)
+        return {
+            field_name: standardize_custom_type(
+                field_name=field_name, value=value, model=self, base_class=AvroRecord, include_type=include_type
+            )
+            for field_name, value in self.asdict().items()
+        }
 
     def serialize(self, serialization_type: str = AVRO) -> bytes:
         """
@@ -48,25 +53,9 @@ class AvroRecord(Record, AvroModel):  # type: ignore
             serialization_type=serialization_type,
         )
 
-    @classmethod
-    def deserialize(
-        cls: typing.Type[CT],
-        data: bytes,
-        serialization_type: str = AVRO,
-        create_instance: bool = True,
-        writer_schema: typing.Optional[typing.Union[JsonDict, typing.Type[CT]]] = None,
-    ) -> typing.Union[JsonDict, CT]:
-        payload = cls.deserialize_to_python(data, serialization_type, writer_schema)
-        obj = cls.parse_obj(payload)
-
-        if not create_instance:
-            return obj.standardize_type()
-        return obj
-
     def to_dict(self) -> JsonDict:
-        return self.standardize_type()
+        return self.standardize_type(include_type=False)
 
     @classmethod
     def _generate_parser(cls: typing.Type[CT]) -> FaustParser:
-        cls._metadata = cls.generate_metadata()
-        return FaustParser(type=cls._klass, metadata=cls._metadata, parent=cls._parent or cls)
+        return FaustParser(type=cls._klass, metadata=cls.get_metadata(), parent=cls._parent or cls)
diff --git a/dataclasses_avroschema/fields/base.py b/dataclasses_avroschema/fields/base.py
index f3b093b..11e60d1 100644
--- a/dataclasses_avroschema/fields/base.py
+++ b/dataclasses_avroschema/fields/base.py
@@ -124,7 +124,7 @@ class Field:
 
     def exist_type(self) -> int:
         # filter by the same field types
-        same_types = [field.type for field in self.parent._user_defined_types if field.type == self.type]
+        same_types = [field.model for field in self.parent._user_defined_types if field.model == self.type]
 
         # If length > 0, means that it is the first appearance
         # of this type, otherwise exist already.
diff --git a/dataclasses_avroschema/fields/fields.py b/dataclasses_avroschema/fields/fields.py
index 0bddd95..1e1c7e6 100644
--- a/dataclasses_avroschema/fields/fields.py
+++ b/dataclasses_avroschema/fields/fields.py
@@ -491,7 +491,7 @@ class EnumField(Field):
         name = metadata.pop("schema_name", self.type.__name__)
 
         if not self.exist_type():
-            user_defined_type = utils.UserDefinedType(name=name, type=self.type)
+            user_defined_type = utils.UserDefinedType(name=name, model=self.type)
             self.parent._user_defined_types.add(user_defined_type)
             return {
                 "type": field_utils.ENUM,
@@ -798,7 +798,7 @@ class RecordField(Field):
         name = alias or metadata.schema_name or self.type.__name__
 
         if not self.exist_type() or alias is not None:
-            user_defined_type = utils.UserDefinedType(name=name, type=self.type)
+            user_defined_type = utils.UserDefinedType(name=name, model=self.type)
             self.parent._user_defined_types.add(user_defined_type)
 
             record_type = self.type.avro_schema_to_python(parent=self.parent)
diff --git a/dataclasses_avroschema/pydantic/main.py b/dataclasses_avroschema/pydantic/main.py
index 0b3175b..9a0e923 100644
--- a/dataclasses_avroschema/pydantic/main.py
+++ b/dataclasses_avroschema/pydantic/main.py
@@ -1,5 +1,5 @@
 import json
-from typing import Any, Callable, Dict, Optional, Type, TypeVar
+from typing import Any, Dict, Type, TypeVar
 
 from fastavro.validation import validate
 
@@ -27,37 +27,43 @@ class AvroBaseModel(BaseModel, AvroModel):  # type: ignore
     def json_schema(cls: Type[CT], *args: Any, **kwargs: Any) -> str:
         return json.dumps(cls.model_json_schema(*args, **kwargs))
 
-    @classmethod
-    def standardize_type(cls: Type[CT], data: dict) -> Any:
+    def _standardize_type(self) -> Dict[str, Any]:
         """
         Standardization factory that converts data according to the
         user-defined pydantic json_encoders prior to passing values
         to the standard type conversion factory
         """
-        for value in data.values():
-            if isinstance(value, dict):
-                cls.standardize_type(value)
+        encoders = self.model_config.get("json_encoders") or {}
+        data = dict(self)
 
-        return standardize_custom_type(data)
+        for k, v in data.items():
+            v_type = type(v)
+            if v_type in encoders:
+                encode_method = encoders[v_type]
+                data[k] = encode_method(v)
+        return data
 
-    def asdict(self, standardize_factory: Optional[Callable[..., Any]] = None) -> JsonDict:
+    def asdict(self) -> JsonDict:
         """
         Returns this model in dictionary form. This method differs from
         pydantic's dict by converting all values to their Avro representation.
         It also doesn't provide the exclude, include, by_alias, etc.
         parameters that dict provides.
         """
-        data = self.model_dump()
-        standardize_method = standardize_factory or self.standardize_type
-
-        # the standardize called can be replaced if we have a custom implementation of asdict
-        # for now I think is better to use the native implementation
-        return standardize_method(data)
+        return {
+            field_name: standardize_custom_type(
+                field_name=field_name, value=field_value, model=self, base_class=AvroBaseModel
+            )
+            for field_name, field_value in self._standardize_type().items()
+        }
 
     @classmethod
     def parse_obj(cls: Type[CT], data: Dict) -> CT:
         return cls.model_validate(obj=data)
 
+    def to_dict(self) -> JsonDict:
+        return self.model_dump()
+
     def serialize(self, serialization_type: str = AVRO) -> bytes:
         """
         Overrides the base AvroModel's serialize method to inject this
@@ -66,7 +72,7 @@ class AvroBaseModel(BaseModel, AvroModel):  # type: ignore
         schema = self.avro_schema_to_python()
 
         return serialization.serialize(
-            self.asdict(standardize_factory=self.standardize_type),
+            self.asdict(),
             schema,
             serialization_type=serialization_type,
         )
@@ -94,5 +100,4 @@ class AvroBaseModel(BaseModel, AvroModel):  # type: ignore
 
     @classmethod
     def _generate_parser(cls: Type[CT]) -> PydanticParser:
-        cls._metadata = cls.generate_metadata()
-        return PydanticParser(type=cls._klass, metadata=cls._metadata, parent=cls._parent or cls)
+        return PydanticParser(type=cls._klass, metadata=cls.get_metadata(), parent=cls._parent or cls)
diff --git a/dataclasses_avroschema/pydantic/v1/main.py b/dataclasses_avroschema/pydantic/v1/main.py
index ae4c72a..8d57eb0 100644
--- a/dataclasses_avroschema/pydantic/v1/main.py
+++ b/dataclasses_avroschema/pydantic/v1/main.py
@@ -1,4 +1,4 @@
-from typing import Any, Callable, Optional, Type, TypeVar
+from typing import Any, Dict, Type, TypeVar
 
 from fastavro.validation import validate
 
@@ -26,38 +26,38 @@ class AvroBaseModel(BaseModel, AvroModel):  # type: ignore
     def json_schema(cls: Type[CT], *args: Any, **kwargs: Any) -> str:
         return cls.schema_json(*args, **kwargs)
 
-    @classmethod
-    def standardize_type(cls: Type[CT], data: dict) -> Any:
+    def _standardize_type(self) -> Dict[str, Any]:
         """
         Standardization factory that converts data according to the
         user-defined pydantic json_encoders prior to passing values
         to the standard type conversion factory
         """
-        encoders = cls.__config__.json_encoders
+        encoders = self.__config__.json_encoders
+        data = dict(self)
+
         for k, v in data.items():
             v_type = type(v)
             if v_type in encoders:
                 encode_method = encoders[v_type]
                 data[k] = encode_method(v)
-            elif isinstance(v, dict):
-                cls.standardize_type(v)
+        return data
 
-        return standardize_custom_type(data)
-
-    def asdict(self, standardize_factory: Optional[Callable[..., Any]] = None) -> JsonDict:
+    def asdict(self) -> JsonDict:
         """
         Returns this model in dictionary form. This method differs from
         pydantic's dict by converting all values to their Avro representation.
         It also doesn't provide the exclude, include, by_alias, etc.
         parameters that dict provides.
         """
-        data = dict(self)
-
-        standardize_method = standardize_factory or self.standardize_type
+        return {
+            field_name: standardize_custom_type(
+                field_name=field_name, value=value, model=self, base_class=AvroBaseModel
+            )
+            for field_name, value in self._standardize_type().items()
+        }
 
-        # the standardize called can be replaced if we have a custom implementation of asdict
-        # for now I think is better to use the native implementation
-        return standardize_method(data)
+    def to_dict(self) -> JsonDict:
+        return dict(self)
 
     def serialize(self, serialization_type: str = AVRO) -> bytes:
         """
@@ -67,7 +67,7 @@ class AvroBaseModel(BaseModel, AvroModel):  # type: ignore
         schema = self.avro_schema_to_python()
 
         return serialization.serialize(
-            self.asdict(standardize_factory=self.standardize_type),
+            self.asdict(),
             schema,
             serialization_type=serialization_type,
         )
@@ -95,5 +95,4 @@ class AvroBaseModel(BaseModel, AvroModel):  # type: ignore
 
     @classmethod
     def _generate_parser(cls: Type[CT]) -> PydanticV1Parser:
-        cls._metadata = cls.generate_metadata()
-        return PydanticV1Parser(type=cls._klass, metadata=cls._metadata, parent=cls._parent or cls)
+        return PydanticV1Parser(type=cls._klass, metadata=cls.get_metadata(), parent=cls._parent or cls)
diff --git a/dataclasses_avroschema/schema_generator.py b/dataclasses_avroschema/schema_generator.py
index 6ea830b..1f9bf7c 100644
--- a/dataclasses_avroschema/schema_generator.py
+++ b/dataclasses_avroschema/schema_generator.py
@@ -2,7 +2,7 @@ import dataclasses
 import inspect
 import json
 from collections import OrderedDict
-from typing import Any, Callable, Dict, List, Optional, Set, Type, TypeVar, Union
+from typing import Any, Dict, List, Optional, Set, Type, TypeVar, Union
 
 from dacite import Config, from_dict
 from fastavro.validation import validate
@@ -12,7 +12,7 @@ from .dacite_config import generate_dacite_config
 from .fields.base import Field
 from .parser import Parser
 from .types import JsonDict
-from .utils import SchemaMetadata, standardize_custom_type
+from .utils import SchemaMetadata, UserDefinedType, standardize_custom_type
 
 AVRO = "avro"
 AVRO_JSON = "avro-json"
@@ -29,7 +29,7 @@ class AvroModel:
     _parser: Optional[Parser] = None
     _klass: Optional[Type] = None
     _metadata: Optional[SchemaMetadata] = None
-    _user_defined_types: Set = set()
+    _user_defined_types: Set[UserDefinedType] = set()
     _parent: Any = None
     _rendered_schema: OrderedDict = dataclasses.field(default_factory=OrderedDict)
 
@@ -40,10 +40,11 @@ class AvroModel:
         return dataclasses.dataclass(cls)
 
     @classmethod
-    def generate_metadata(cls: "Type[CT]") -> SchemaMetadata:
-        meta = getattr(cls._klass, "Meta", type)
-
-        return SchemaMetadata.create(meta)
+    def get_metadata(cls: "Type[CT]") -> SchemaMetadata:
+        if cls._metadata is None:
+            meta = getattr(cls._klass, "Meta", type)
+            cls._metadata = SchemaMetadata.create(meta)
+        return cls._metadata
 
     @classmethod
     def generate_schema(cls: "Type[CT]", schema_type: str = "avro") -> Optional[OrderedDict]:
@@ -62,10 +63,22 @@ class AvroModel:
 
         return cls._rendered_schema
 
+    @classmethod
+    def get_user_defined_type(cls, *, name: str) -> Optional[Type]:
+        """
+        Attributes:
+            name str: Model name to be search in the user_defined_types set
+
+        Returns:
+            Model CT
+        """
+        return next(
+            (user_type.model for user_type in cls._user_defined_types if user_type.model.__name__ == name), None
+        )
+
     @classmethod
     def _generate_parser(cls: "Type[CT]") -> Parser:
-        cls._metadata = cls.generate_metadata()
-        return Parser(type=cls._klass, metadata=cls._metadata, parent=cls._parent or cls)
+        return Parser(type=cls._klass, metadata=cls.get_metadata(), parent=cls._parent or cls)
 
     @classmethod
     def avro_schema(cls: "Type[CT]", case_type: Optional[str] = None, **kwargs) -> str:
@@ -115,27 +128,6 @@ class AvroModel:
         cls._parser = None
         cls._parent = None
 
-    def asdict(self, standardize_factory: Optional[Callable[..., Any]] = None) -> JsonDict:
-        if standardize_factory is not None:
-            return dataclasses.asdict(
-                self,
-                dict_factory=lambda x: {key: standardize_factory(value) for key, value in x},
-            )  # type: ignore
-        return dataclasses.asdict(self)  # type: ignore
-
-    def serialize(self, serialization_type: str = AVRO) -> bytes:
-        klass = type(self)
-        schema = _schemas_cache.get(klass)
-        if schema is None:
-            schema = self.avro_schema_to_python()
-            _schemas_cache[klass] = schema
-
-        return serialization.serialize(
-            self.asdict(standardize_factory=standardize_custom_type),
-            schema,
-            serialization_type=serialization_type,
-        )
-
     @classmethod
     def deserialize(
         cls: "Type[CT]",
@@ -148,7 +140,7 @@ class AvroModel:
         obj = cls.parse_obj(payload)
 
         if not create_instance:
-            return obj.asdict()
+            return obj.to_dict()
         return obj
 
     @classmethod
@@ -167,8 +159,9 @@ class AvroModel:
             schema = cls.avro_schema_to_python()
             _schemas_cache[cls] = schema
         return serialization.deserialize(
-            data,
-            schema,
+            data=data,
+            schema=schema,
+            model=cls,
             serialization_type=serialization_type,
             writer_schema=writer_schema,  # type: ignore
         )
@@ -181,17 +174,6 @@ class AvroModel:
             _dacite_config_cache[cls] = config
         return from_dict(data_class=cls, data=data, config=config)
 
-    def validate(self) -> bool:
-        schema = self.avro_schema_to_python()
-        return validate(self.asdict(), schema)
-
-    def to_dict(self) -> JsonDict:
-        return self.asdict()
-
-    def to_json(self, **kwargs: Any) -> str:
-        data = serialization.to_json(self.asdict())
-        return json.dumps(data, **kwargs)
-
     @classmethod
     def fake(cls: "Type[CT]", **data: Any) -> CT:
         """
@@ -206,3 +188,35 @@ class AvroModel:
         payload.update(data)
 
         return from_dict(data_class=cls, data=payload, config=generate_dacite_config(cls))
+
+    def asdict(self) -> JsonDict:
+        return {
+            field.name: standardize_custom_type(
+                field_name=field.name, value=getattr(self, field.name), model=self, base_class=AvroModel
+            )
+            for field in dataclasses.fields(self)  # type: ignore
+        }
+
+    def serialize(self, serialization_type: str = AVRO) -> bytes:
+        klass = type(self)
+        schema = _schemas_cache.get(klass)
+        if schema is None:
+            schema = self.avro_schema_to_python()
+            _schemas_cache[klass] = schema
+
+        return serialization.serialize(
+            self.asdict(),
+            schema,
+            serialization_type=serialization_type,
+        )
+
+    def validate(self) -> bool:
+        schema = self.avro_schema_to_python()
+        return validate(self.asdict(), schema)
+
+    def to_dict(self) -> JsonDict:
+        return dataclasses.asdict(self)  # type: ignore
+
+    def to_json(self, **kwargs: Any) -> str:
+        data = serialization.to_json(self.to_dict())
+        return json.dumps(data, **kwargs)
diff --git a/dataclasses_avroschema/serialization.py b/dataclasses_avroschema/serialization.py
index f0c3116..a6c4a50 100644
--- a/dataclasses_avroschema/serialization.py
+++ b/dataclasses_avroschema/serialization.py
@@ -9,6 +9,9 @@ import fastavro
 
 from .types import JsonDict
 
+if typing.TYPE_CHECKING:
+    from .schema_generator import CT
+
 DATETIME_STR_FORMAT = "%Y-%m-%dT%H:%M:%S%z"
 DATE_STR_FORMAT = "%Y-%m-%d"
 TIME_STR_FORMAT = "%H:%M:%S"
@@ -36,11 +39,13 @@ def serialize(payload: typing.Dict, schema: typing.Dict, serialization_type: str
 
 
 def deserialize(
+    *,
     data: bytes,
-    schema: typing.Dict,
+    schema: JsonDict,
+    model: "CT",
     serialization_type: str = "avro",
     writer_schema: typing.Optional[JsonDict] = None,
-) -> typing.Dict:
+) -> JsonDict:
     if serialization_type == "avro":
         input_stream: typing.Union[io.BytesIO, io.StringIO] = io.BytesIO(data)
 
@@ -48,6 +53,8 @@ def deserialize(
             input_stream,
             writer_schema=writer_schema or schema,
             reader_schema=schema,
+            return_record_name=True,
+            return_record_name_override=True,
         )
 
     elif serialization_type == "avro-json":
@@ -61,7 +68,31 @@ def deserialize(
 
     input_stream.flush()
 
-    return payload  # type: ignore
+    return sanitize_unions(data=payload, model=model)  # type: ignore
+
+
+def sanitize_unions(*, data: JsonDict, model: "CT") -> JsonDict:
+    """
+    This function tries to convert cast all the cases that have
+    `unions` with a Tuple format (AvroType, payload), for example
+    (Car, {"brand": "fiat"}). This cases happens when there are
+    avro unions of records which are similar or equals among them
+    """
+    cleaned_data = {}
+    for field_name, field_value in data.items():
+        if isinstance(field_value, dict):
+            field_value = sanitize_unions(data=field_value, model=model)
+        elif isinstance(field_value, tuple) and len(field_value) == 2:
+            # the first value is the model/record name and the second
+            # is its payload
+            model_name, model_dict_value = field_value
+            avro_model = model.get_user_defined_type(name=model_name)
+            if avro_model is not None:
+                field_value = avro_model.parse_obj(model_dict_value)
+
+        cleaned_data[field_name] = field_value
+
+    return cleaned_data
 
 
 def datetime_to_str(value: datetime.datetime) -> str:
diff --git a/dataclasses_avroschema/utils.py b/dataclasses_avroschema/utils.py
index 93cac67..924eec0 100644
--- a/dataclasses_avroschema/utils.py
+++ b/dataclasses_avroschema/utils.py
@@ -6,7 +6,10 @@ from functools import lru_cache
 
 from typing_extensions import Annotated, get_origin
 
-from .types import FieldInfo, JsonDict
+from .types import FieldInfo, JsonDict, UnionType
+
+if typing.TYPE_CHECKING:
+    from .schema_generator import AvroModel
 
 try:
     import pydantic  # pragma: no cover
@@ -22,20 +25,20 @@ except ImportError:  # type: ignore # pragma: no cover
 
 
 @lru_cache(maxsize=None)
-def is_pydantic_model(klass: type) -> bool:
+def is_pydantic_model(klass: typing.Type) -> bool:
     if pydantic is not None:
         return issubclass(klass, v1.BaseModel) or issubclass(klass, pydantic.BaseModel)
     return False
 
 
 @lru_cache(maxsize=None)
-def is_faust_record(klass: type) -> bool:
+def is_faust_record(klass: typing.Type) -> bool:
     if faust is not None:
         return issubclass(klass, faust.Record)
     return False
 
 
-def is_union(a_type: type) -> bool:
+def is_union(a_type: typing.Type) -> bool:
     """
     Given a python type, return True if is typing.Union, otherwise False
 
@@ -45,10 +48,16 @@ def is_union(a_type: type) -> bool:
     Returns:
         bool
     """
-    return isinstance(a_type, typing._GenericAlias) and a_type.__origin__ is typing.Union  # type: ignore
+    if UnionType is None:
+        return (
+            isinstance(a_type, typing._GenericAlias) and a_type.__origin__ is typing.Union  # type: ignore
+        )
+    return (
+        isinstance(a_type, typing._GenericAlias) and a_type.__origin__ is typing.Union or isinstance(a_type, UnionType)  # type: ignore
+    )
 
 
-def is_self_referenced(a_type: type, parent: type) -> bool:
+def is_self_referenced(a_type: typing.Type, parent: typing.Type) -> bool:
     """
     Given a python type, return True if is self referenced, meaning
     that is instance of typing.ForwardRef, otherwise False
@@ -70,30 +79,53 @@ def is_self_referenced(a_type: type, parent: type) -> bool:
     return (
         isinstance(a_type, typing._GenericAlias)  # type: ignore
         and a_type.__args__
-        and isinstance(a_type.__args__[0], typing.ForwardRef)  # type: ignore
+        and isinstance(a_type.__args__[0], typing.ForwardRef)
     ) or a_type == parent
 
 
-def is_annotated(a_type: typing.Any) -> bool:
+def is_annotated(a_type: typing.Type) -> bool:
     origin = get_origin(a_type)
     return origin is not None and isinstance(origin, type) and issubclass(origin, Annotated)  # type: ignore[arg-type]
 
 
-def rebuild_annotation(a_type: typing.Any, field_info: FieldInfo) -> typing.Type:
+def rebuild_annotation(a_type: typing.Type, field_info: FieldInfo) -> typing.Type:
     return Annotated[a_type, field_info]  # type: ignore[return-value]
 
 
-def standardize_custom_type(value: typing.Any) -> typing.Any:
+def standardize_custom_type(
+    *,
+    field_name: str,
+    value: typing.Any,
+    model: "AvroModel",
+    base_class: typing.Type["AvroModel"],
+    include_type: bool = True,
+) -> typing.Any:
     if isinstance(value, dict):
-        return {k: standardize_custom_type(v) for k, v in value.items()}
+        return {
+            k: standardize_custom_type(field_name=field_name, value=v, model=model, base_class=base_class)
+            for k, v in value.items()
+        }
     elif isinstance(value, list):
-        return [standardize_custom_type(v) for v in value]
+        return [
+            standardize_custom_type(field_name=field_name, value=v, model=model, base_class=base_class) for v in value
+        ]
     elif isinstance(value, tuple):
-        return tuple(standardize_custom_type(v) for v in value)
+        return tuple(
+            standardize_custom_type(field_name=field_name, value=v, model=model, base_class=base_class) for v in value
+        )
     elif isinstance(value, enum.Enum):
         return value.value
-    elif is_pydantic_model(type(value)) or is_faust_record(type(value)):  # type: ignore[arg-type]
-        return standardize_custom_type(value.asdict())
+    elif isinstance(value, base_class):
+        if is_faust_record(type(value)):  # type: ignore[arg-type]
+            # we need to do a trick because we can not overrride asdict from faust..
+            # once the function interface is introduced we can remove this check
+            asdict = value.standardize_type()  # type: ignore
+        else:
+            asdict = value.asdict()
+
+        if is_union(model.__annotations__[field_name]) and include_type:
+            asdict["-type"] = value.__class__.__name__
+        return asdict
 
     return value
 
@@ -102,7 +134,7 @@ def standardize_custom_type(value: typing.Any) -> typing.Any:
 class SchemaMetadata:
     schema_name: typing.Optional[str] = None
     schema_doc: typing.Union[bool, str] = True
-    namespace: typing.Optional[typing.List[str]] = None
+    namespace: typing.Optional[str] = None
     aliases: typing.Optional[typing.List[str]] = None
     alias_nested_items: typing.Dict[str, str] = dataclasses.field(default_factory=dict)
     dacite_config: typing.Optional[JsonDict] = None
@@ -157,7 +189,7 @@ class FieldMetadata:
 
 class UserDefinedType(typing.NamedTuple):
     name: str
-    type: typing.Any
+    model: typing.Type
 
 
 epoch: datetime = datetime(1970, 1, 1, tzinfo=timezone.utc)
diff --git a/docs/records.md b/docs/records.md
index 8a3179e..5b96b9a 100644
--- a/docs/records.md
+++ b/docs/records.md
@@ -77,6 +77,8 @@ class Meta:
 
 `convert_literal_to_enum Optional[bool]`: Whether convert `Literal string` to `enum`
 
+`dacite_config Optional[Dict]`: Dacite custom config
+
 ## Record to json and dict
 
 You can get the `json` and `dict` representation of your instance using `to_json` and `to_dict` methods:
@@ -142,93 +144,8 @@ print(bus)
 
 *(This script is complete, it should run "as is")*
 
-### Override dacite config
-
-There are some use cases where a custom `dacite` config is needed, so you can provide one using the `dacite_config` in the `class Meta`.
-
-For example using [Strict unions match](https://github.com/konradhalas/dacite#strict-unions-match)
-
-=== "Default dacite configuration"
-
-    ```python
-    import typing
-    from dataclasses import dataclass
-
-    from dataclasses_avroschema import AvroModel
-
-
-    @dataclass
-    class Car(AvroModel):
-        total: int
-
-
-    @dataclass
-    class Bus(AvroModel):
-        driver: str
-        total: int
-
-
-    @dataclass
-    class Trip(AvroModel):
-        transport: typing.Union[Car, Bus]
-
-    data = {"driver": "Marcos", "total": 10}
-    bus = Bus.parse_obj(data=data)
-
-    serialized_val = Trip(transport=bus).serialize()
-
-    print(Trip.deserialize(serialized_val, create_instance=False)) 
-    # >>> {"transport": {"driver": "Marcos", "total": 10}}
-    
-    instance = Trip.deserialize(serialized_val)
-    print(instance.transport)  # This is a Car but it should be a Bus!!!
-    # >>> Car(total=10)
-    ```
-
-=== "Custom dacite configuration"
-
-    ```python
-    import typing
-    from dataclasses import dataclass
-
-    from dataclasses_avroschema import AvroModel
-
-
-    @dataclass
-    class Car(AvroModel):
-        total: int
-
-
-    @dataclass
-    class Bus(AvroModel):
-        driver: str
-        total: int
-
-
-    @dataclass
-    class Trip(AvroModel):
-        transport: typing.Union[Car, Bus]
-
-        class Meta:
-            dacite_config = {
-                "strict_unions_match": True,
-                "strict": True,
-            }
-
-    data = {"driver": "Marcos", "total": 10}
-    bus = Bus.parse_obj(data=data)
-
-    serialized_val = Trip(transport=bus).serialize()
-    print(Trip.deserialize(serialized_val, create_instance=False))
-    # >>> {"transport": {"driver": "Marcos", "total": 10}}
-    
-    instance = Trip.deserialize(serialized_val)
-    print(instance.transport)  # Is it s Bus and not a Car!!!
-    # >>> Bus(driver='Marcos', total=10)
-
-    ```
-
-*(This script is complete, it should run "as is")*
+!!! note
+    There are some use cases where a custom [dacite](https://github.com/konradhalas/dacite) config is needed, so you can provide one using the `dacite_config` in the `class Meta`
 
 ## Validation
 


Test patch:
diff --git a/tests/schemas/pydantic/test_pydantic.py b/tests/schemas/pydantic/test_pydantic.py
index 5b3befb..2232b6d 100644
--- a/tests/schemas/pydantic/test_pydantic.py
+++ b/tests/schemas/pydantic/test_pydantic.py
@@ -569,12 +569,11 @@ def test_exclude_field() -> None:
     assert Message.avro_schema()
 
     message = Message(internal_field="internal", public_field="public")
-    assert "internal_field" not in message.model_dump()
-
-    with pytest.raises(ValueError) as excinfo:
-        message.serialize(serialization_type="avro-json")
+    assert "internal_field" not in message.to_dict()
 
-    assert str(excinfo.value) == "no value and no default for internal_field"
+    event = message.serialize(serialization_type="avro-json")
+    assert Message.deserialize(event, serialization_type="avro-json") == message
+    assert Message.deserialize(event, serialization_type="avro-json", create_instance=False) == message.model_dump()
 
 
 def test_exclude_field_with_default() -> None:
diff --git a/tests/serialization/test_model_utils.py b/tests/serialization/test_model_utils.py
index c2f8429..c58d49c 100644
--- a/tests/serialization/test_model_utils.py
+++ b/tests/serialization/test_model_utils.py
@@ -5,7 +5,7 @@ from dataclasses import dataclass
 
 import pytest
 
-from dataclasses_avroschema import AvroModel
+from dataclasses_avroschema import AvroModel, schema_generator
 from tests.serialization.test_serialization import CLASSES_DATA_BINARY
 
 
@@ -43,9 +43,9 @@ def test_dacite_config():
 
     # It matches a Car object because it is the first element in the `Union`
     # In order to match the proper element use `strict_unions_match` (check next test)
-    assert Trip.deserialize(serialized_val, create_instance=False) == {"transport": {"total": 10}}
+    assert Trip.deserialize(serialized_val, create_instance=False) == {"transport": {"driver": "Marcos", "total": 10}}
     instance = Trip.deserialize(serialized_val)
-    assert instance.transport == Car(total=10)
+    assert instance.transport == bus
 
 
 def test_custom_dacite_config():
@@ -94,3 +94,8 @@ def test_custom_dacite_config():
     }
     instance = Trip.deserialize(serialized_val)
     assert instance.transport == bus
+
+    dacite_config = schema_generator._dacite_config_cache[Trip]
+    assert dacite_config.strict_unions_match
+    assert dacite_config.strict
+    assert dacite_config.cast != []
diff --git a/tests/serialization/test_nested_schema_serialization.py b/tests/serialization/test_nested_schema_serialization.py
index e393d6d..2610712 100644
--- a/tests/serialization/test_nested_schema_serialization.py
+++ b/tests/serialization/test_nested_schema_serialization.py
@@ -7,10 +7,16 @@ import pytest
 from dataclasses_avroschema import AvroModel
 from dataclasses_avroschema.faust import AvroRecord
 from dataclasses_avroschema.pydantic import AvroBaseModel
+from dataclasses_avroschema.pydantic.v1 import AvroBaseModel as AvroBaseModelV1
 
 parametrize_base_model = pytest.mark.parametrize(
     "model_class, decorator",
-    [(AvroModel, dataclasses.dataclass), (AvroBaseModel, lambda f: f), (AvroRecord, lambda f: f)],
+    [
+        (AvroModel, dataclasses.dataclass),
+        (AvroBaseModel, lambda f: f),
+        (AvroBaseModelV1, lambda f: f),
+        (AvroRecord, lambda f: f),
+    ],
 )
 
 
@@ -242,8 +248,8 @@ def test_nested_schemas_splitted_with_unions(model_class: typing.Type[AvroModel]
     class A(model_class):
         s: typing.Union[S1, S2]
 
-        class Meta:
-            namespace = "namespace"
+        # class Meta:
+        #     namespace = "namespace"
 
     @decorator
     class B(model_class):
@@ -307,3 +313,35 @@ def test_nested_several_layers(model_class: typing.Type[AvroModel], decorator: t
     user = User(name="Alex", friends=[Friend(name="Mr. Robot", hobbies=["fishing", "codding"])])
 
     assert User.deserialize(user.serialize()) == user
+
+
+@parametrize_base_model
+def test_union_with_multiple_records(model_class: typing.Type[AvroModel], decorator: typing.Callable):
+    @decorator
+    class EventOne(model_class):
+        name: str
+        tag: typing.Literal["EventOne"] = "EventOne"
+
+    @decorator
+    class EventTwo(model_class):
+        name: str
+        tag: typing.Literal["EventTwo"] = "EventTwo"
+
+    @decorator
+    class EventManager(model_class):
+        event: typing.Union[EventOne, EventTwo]
+        capacity: int = 100
+
+    # check union with first element
+    event = EventManager(event=EventOne(name="hello Event one"))
+    event_serialized = event.serialize()
+
+    assert event_serialized == b"\x00\x1ehello Event one\x10EventOne\xc8\x01"
+    assert EventManager.deserialize(event_serialized) == event
+
+    # check union with second element
+    event = EventManager(event=EventTwo(name="hello Event two"), capacity=150)
+    event_serialized = event.serialize()
+
+    assert event_serialized == b"\x02\x1ehello Event two\x10EventTwo\xac\x02"
+    assert EventManager.deserialize(event_serialized) == event