Repository: marcosschroh/dataclasses-avroschema

Issue:
Name Collision for certain relationship even when using namespace
**Describe the bug**
The avro serialization with `fastavro` for three classes
  - `A`,
  - `B` (depending on `A`),
  - `C` (depending on `A` and `B`),

fails with

```
fastavro._schema_common.SchemaParseException: redefined named type: namespace.A
```

even when using a namespace for `A`.

**To Reproduce**
(*EDITED:* changed from using `dataclasses_avroschema.avrodantic.AvroBaseModel` to `dataclasses_avroschema.AvroModel`.)
```python
from dataclasses import dataclass
from dataclasses_avroschema import AvroModel


@dataclass
class A(AvroModel):
    class Meta:
        namespace = "namespace"


@dataclass
class B(AvroModel):
    a: A


@dataclass
class C(AvroModel):
    # class Meta:
    #     alias_nested_items = {"a": "A2"}  # Workaround

    b: B
    a: A


if __name__ == "__main__":
    b = B(a=A())
    c = C(b=B(a=A()), a=A())

    b.serialize()
    c.serialize()
```


**Expected behavior**
The code above should run but fails with the error (i used `dataclasses-avroschema==0.30.3`).
```
Traceback (most recent call last):
  File "/home/tbartsch/test/avro_issue.py", line 28, in <module>
    c.serialize()
  File "/home/tbartsch/test/venv/lib/python3.10/site-packages/dataclasses_avroschema/schema_generator.py", line 144, in serialize
    return serialize(self.asdict(), schema, serialization_type=serialization_type)
  File "/home/tbartsch/test/venv/lib/python3.10/site-packages/dataclasses_avroschema/serialization.py", line 20, in serialize
    fastavro.schemaless_writer(file_like_output, schema, payload)
  File "fastavro/_write.pyx", line 785, in fastavro._write.schemaless_writer
  File "fastavro/_schema.pyx", line 118, in fastavro._schema.parse_schema
  File "fastavro/_schema.pyx", line 257, in fastavro._schema._parse_schema
  File "fastavro/_schema.pyx", line 302, in fastavro._schema.parse_field
  File "fastavro/_schema.pyx", line 202, in fastavro._schema._parse_schema
  File "fastavro/_schema.pyx", line 249, in fastavro._schema._parse_schema
fastavro._schema_common.SchemaParseException: redefined named type: namespace.A

Process finished with exit code 1
```


Discussion:
github-actions[bot]: [PR Preview Action](https://github.com/rossjrw/pr-preview-action) v1.1.1
:---:
:rocket: Deployed preview to https://marcosschroh.github.io/dataclasses-avroschema/pr-preview/pr-198/
on branch [`gh-pages`](https://github.com/marcosschroh/dataclasses-avroschema/tree/gh-pages) at 2022-09-30 18:50 UTC
<!-- Sticky Pull Request Commentpr-preview -->

Patch:
diff --git a/dataclasses_avroschema/fields.py b/dataclasses_avroschema/fields.py
index 46fc999..dee3f84 100644
--- a/dataclasses_avroschema/fields.py
+++ b/dataclasses_avroschema/fields.py
@@ -817,9 +817,9 @@ class RecordField(BaseField):
 
         if not self.exist_type() or alias is not None:
             user_defined_type = utils.UserDefinedType(name=name, type=self.type)
-            self.parent.user_defined_types += (user_defined_type,)
+            self.parent.user_defined_types.add(user_defined_type)
 
-            record_type = self.type.avro_schema_to_python(root=self.parent)
+            record_type = self.type.avro_schema_to_python(parent=self.parent)
             record_type["name"] = name
         else:
             if metadata.namespace is None:
diff --git a/dataclasses_avroschema/schema_generator.py b/dataclasses_avroschema/schema_generator.py
index eeeeb9a..b80c7d6 100644
--- a/dataclasses_avroschema/schema_generator.py
+++ b/dataclasses_avroschema/schema_generator.py
@@ -3,7 +3,7 @@ import enum
 import inspect
 import json
 from collections import OrderedDict
-from typing import Any, Dict, List, Optional, Tuple, 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
@@ -26,8 +26,8 @@ class AvroModel:
     schema_def: Optional[AvroSchemaDefinition] = None
     klass: Optional[Type] = None
     metadata: Optional[SchemaMetadata] = None
-    user_defined_types: Tuple = ()
-    root: Any = None
+    user_defined_types: Set = set()
+    parent: Any = None
     rendered_schema: OrderedDict = dataclasses.field(default_factory=OrderedDict)
 
     @classmethod
@@ -37,7 +37,7 @@ class AvroModel:
         return dataclasses.dataclass(cls)
 
     @classmethod
-    def generate_metadata(cls: Any) -> SchemaMetadata:
+    def generate_metadata(cls: Type[CT]) -> SchemaMetadata:
         meta = getattr(cls.klass, "Meta", type)
 
         return SchemaMetadata.create(meta)
@@ -63,24 +63,27 @@ class AvroModel:
     def _generate_avro_schema(cls: Type[CT]) -> AvroSchemaDefinition:
         metadata = cls.generate_metadata()
         cls.metadata = metadata
-        return AvroSchemaDefinition("record", cls.klass, metadata=metadata, parent=cls.root or cls)
+        return AvroSchemaDefinition("record", cls.klass, metadata=metadata, parent=cls.parent or cls)
 
     @classmethod
     def avro_schema(cls: Type[CT], case_type: Optional[str] = None) -> str:
         avro_schema = cls.generate_schema(schema_type=AVRO)
 
-        # After generating the avro schema, reset the raw_fields to the init
-        cls.user_defined_types = ()
-
         if case_type is not None:
             avro_schema = case.case_record(cls.rendered_schema, case_type)  # type: ignore
 
         return json.dumps(avro_schema)
 
     @classmethod
-    def avro_schema_to_python(cls: Any, root: Any = None) -> Dict[str, Any]:
-        if root is not None:
-            cls.root = root
+    def avro_schema_to_python(cls: Type[CT], parent: Optional["AvroModel"] = None) -> Dict[str, Any]:
+        if parent is not None:
+            # in this case the current class is a child with a parent
+            # we recalculate the schema definition to prevent re usages
+            cls.parent = parent
+            cls.schema_def = None
+        else:
+            # in this case the current class is a parent so we reset the user_defined_types
+            cls.user_defined_types = set()
 
         return json.loads(cls.avro_schema())
 


Test patch:
diff --git a/tests/schemas/test_fastavro_paser_schema.py b/tests/schemas/test_fastavro_paser_schema.py
index b233656..ec95587 100644
--- a/tests/schemas/test_fastavro_paser_schema.py
+++ b/tests/schemas/test_fastavro_paser_schema.py
@@ -438,3 +438,27 @@ def test_two_different_child_records():
             namespace = "test.namespace"
 
     assert parse_schema(HolidayAlbum.avro_schema_to_python())
+
+
+def test_nested_schemas_splitted() -> None:
+    """
+    This test will cover the cases when nested schemas are
+    used in a separate way.
+    """
+
+    class A(AvroModel):
+        class Meta:
+            namespace = "namespace"
+
+    class B(AvroModel):
+        a: A
+
+    class C(AvroModel):
+        b: B
+        a: A
+
+    # first the B schema is generated
+    assert parse_schema(B.avro_schema_to_python())
+
+    # then check that the C schema is valid
+    assert parse_schema(C.avro_schema_to_python())
diff --git a/tests/serialization/test_nested_schema_serialization.py b/tests/serialization/test_nested_schema_serialization.py
index 8c447da..96d8d0b 100644
--- a/tests/serialization/test_nested_schema_serialization.py
+++ b/tests/serialization/test_nested_schema_serialization.py
@@ -154,3 +154,67 @@ def test_one_to_many_map_relationship():
     # assert User.deserialize(avro_json_binary) == user
 
     assert user.to_dict() == expected
+
+
+def test_nested_schemas_splitted() -> None:
+    """
+    This test will cover the cases when nested schemas are
+    used in a separate way.
+    """
+
+    @dataclasses.dataclass
+    class A(AvroModel):
+        class Meta:
+            namespace = "namespace"
+
+    @dataclasses.dataclass
+    class B(AvroModel):
+        a: A
+
+    @dataclasses.dataclass
+    class C(AvroModel):
+        b: B
+        a: A
+
+    b = B(a=A())
+    c = C(b=B(a=A()), a=A())
+
+    assert b.serialize() == b""
+    assert c.serialize() == b""
+
+
+def test_nested_schemas_splitted_with_unions() -> None:
+    """
+    This test will cover the cases when nested schemas with Unions that are
+    used in a separate way.
+    """
+
+    @dataclasses.dataclass
+    class S1(AvroModel):
+        pass
+
+    @dataclasses.dataclass
+    class S2(AvroModel):
+        pass
+
+    @dataclasses.dataclass
+    class A(AvroModel):
+        s: typing.Union[S1, S2]
+
+        class Meta:
+            namespace = "namespace"
+
+    @dataclasses.dataclass
+    class B(AvroModel):
+        a: A
+
+    @dataclasses.dataclass
+    class C(AvroModel):
+        b: B
+        a: A
+
+    b = B(a=A(s=S1()))
+    c = C(b=B(a=A(s=S1())), a=A(s=S1()))
+
+    assert b.serialize() == b"\x00"
+    assert c.serialize() == b"\x00\x00"