Repository: fastavro/fastavro

Issue:
RecursionError: maximum recursion depth exceeded when trying to serialize payload with a schema that has recurse relationships using avro json
It is not possible to `serialize` payloads with a schema that have recursive relationships using `avro-json`. When using `avro binary` it works as expected.

- python version: `all`
- fastavro: `1.9.4`

To reproduce we use a schema with a recursive relationship: 

```python
schema = {
    'type': 'record', 
    'name': 'User', 
    'fields': [
        {'name': 'name', 'type': 'string'}, 
        {'name': 'age', 'type': 'long'}, 
        {'name': 'friend', 'type': ['null', 'User'], 'default': None}
    ]
}

fastavro.parse_schema(schema=schema) # This works as expected!!

data = {
    "name": "juan",
    "age": 20,
    "friend": {
        "name": "john",
        "age": 20,
    },
}
```

Then when trying to serialize:

```python
import io
import typing

import fastavro


# Helpers from fastavro
def serialize(*, payload: typing.Dict, schema: typing.Dict) -> bytes:
    file_like_output = io.StringIO()
    fastavro.json_writer(file_like_output, schema, [payload])
    value = file_like_output.getvalue().encode("utf-8")

    file_like_output.flush()

    return value

serialize(payload=data, schema=schema)

...
  File "/Users/Projects/.venv/lib/python3.10/site-packages/fastavro/io/parser.py", line 70, in _parse
    symbols.append(self._parse(candidate_schema))
  File "/Users/Projects/.venv/lib/python3.10/site-packages/fastavro/io/parser.py", line 120, in _parse
    return self._parse(self.named_schemas[record_type])
  File "/Users/Projects/.venv/lib/python3.10/site-packages/fastavro/io/parser.py", line 58, in _parse
    0, self._parse(field["type"], field.get("default", NO_DEFAULT))
  File "/Users/Projects/.venv/lib/python3.10/site-packages/fastavro/io/parser.py", line 106, in _parse
    return String(default=default)
RecursionError: maximum recursion depth exceeded
```

If we try `avro binary` then it works as expected:

```python
import io
import typing

import fastavro


def serialize(*, payload: typing.Dict, schema: typing.Dict) -> bytes:
    file_like_output = io.BytesIO()
    fastavro.schemaless_writer(file_like_output, schema, payload)
    value = file_like_output.getvalue()
    file_like_output.flush()

    return value

print(serialize(payload=data, schema=schema))

>>> b'\x08juan(\x02\x08john(\x00
```

Discussion:
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/fastavro/fastavro/pull/773?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=fastavro) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 55.05%. Comparing base [(`530c81b`)](https://app.codecov.io/gh/fastavro/fastavro/commit/530c81b33398cb1062f848e0ab296b6efcae9972?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=fastavro) to head [(`68ab16a`)](https://app.codecov.io/gh/fastavro/fastavro/pull/773?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=fastavro).
> Report is 2 commits behind head on master.


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


```diff
@@             Coverage Diff             @@
##           master     #773       +/-   ##
===========================================
- Coverage   98.51%   55.05%   -43.47%     
===========================================
  Files          32       32               
  Lines        2292     2305       +13     
===========================================
- Hits         2258     1269      -989     
- Misses         34     1036     +1002     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/fastavro/fastavro/pull/773?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=fastavro).   
: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=fastavro).

scottbelden: Thanks. Will try to take a look soon

Patch:
diff --git a/fastavro/io/parser.py b/fastavro/io/parser.py
index d518170..a918871 100644
--- a/fastavro/io/parser.py
+++ b/fastavro/io/parser.py
@@ -35,6 +35,7 @@ from ..schema import extract_record_type
 class Parser:
     def __init__(self, schema, named_schemas, action_function):
         self.schema = schema
+        self._processed_records = []
         self.named_schemas = named_schemas
         self.action_function = action_function
         self.stack = self.parse()
@@ -45,20 +46,45 @@ class Parser:
         root.production.insert(0, root)
         return [root, symbol]
 
+    def _process_record(self, schema, default, schema_name=None):
+        production = []
+
+        production.append(RecordStart(default=default))
+        for field in schema["fields"]:
+            field_name = field["name"]
+            production.insert(0, FieldStart(field_name))
+
+            if schema_name is not None and schema_name in field["type"]:
+                # this meanns a recursive relationship, so we force a `null`
+                internal_record = Sequence(
+                    Alternative([Null()], ["null"], default=None), Union()
+                )
+            else:
+
+                internal_record = self._parse(
+                    field["type"], field.get("default", NO_DEFAULT)
+                )
+
+            production.insert(0, internal_record)
+            production.insert(0, FieldEnd())
+        production.insert(0, RecordEnd())
+
+        return production
+
     def _parse(self, schema, default=NO_DEFAULT):
         record_type = extract_record_type(schema)
 
         if record_type == "record":
             production = []
+            schema_name = schema["name"]
 
-            production.append(RecordStart(default=default))
-            for field in schema["fields"]:
-                production.insert(0, FieldStart(field["name"]))
-                production.insert(
-                    0, self._parse(field["type"], field.get("default", NO_DEFAULT))
+            if schema_name not in self._processed_records:
+                self._processed_records.append(schema_name)
+                production = self._process_record(schema, default)
+            else:
+                production = self._process_record(
+                    schema, default, schema_name=schema_name
                 )
-                production.insert(0, FieldEnd())
-            production.insert(0, RecordEnd())
 
             seq = Sequence(*production)
             return seq


Test patch:
diff --git a/tests/test_json.py b/tests/test_json.py
index 7680ffe..90bc09d 100644
--- a/tests/test_json.py
+++ b/tests/test_json.py
@@ -425,6 +425,103 @@ def test_with_dependent_schema():
     assert records == new_records
 
 
+def test_self_one_to_one_relationship():
+    """Tests a schema with self dependent schema
+    https://github.com/fastavro/fastavro/issues/765"""
+
+    schema = {
+        "type": "record",
+        "name": "User",
+        "fields": [
+            {"name": "name", "type": "string"},
+            {"name": "age", "type": "long"},
+            {"name": "friend", "type": ["null", "User"], "default": None},
+        ],
+    }
+
+    records = [
+        {
+            "name": "juan",
+            "age": 20,
+            "friend": {
+                "name": "john",
+                "age": 30,
+                "friend": None,
+            },
+        }
+    ]
+
+    new_records = roundtrip(schema, records)
+    assert records == new_records
+
+
+def test_multiple_self_one_to_one_relationship():
+    """Tests a schema with self dependent schema
+    https://github.com/fastavro/fastavro/issues/765"""
+
+    schema = {
+        "type": "record",
+        "name": "User",
+        "fields": [
+            {"name": "name", "type": "string"},
+            {"name": "age", "type": "long"},
+            {"name": "friend", "type": ["null", "User"], "default": None},
+            {"name": "colleague", "type": ["null", "User"], "default": None},
+        ],
+    }
+
+    records = [
+        {
+            "name": "juan",
+            "age": 20,
+            "friend": {
+                "name": "john",
+                "age": 30,
+                "friend": None,
+                "colleague": None,
+            },
+            "colleague": {
+                "name": "marcos",
+                "age": 35,
+                "friend": None,
+                "colleague": None,
+            },
+        }
+    ]
+
+    new_records = roundtrip(schema, records)
+    assert records == new_records
+
+
+def test_self_one_to_one_relationship_null_required():
+    """Tests a schema with self dependent schema without `null`
+    https://github.com/fastavro/fastavro/issues/765"""
+
+    schema = {
+        "type": "record",
+        "name": "User",
+        "fields": [
+            {"name": "name", "type": "string"},
+            {"name": "age", "type": "long"},
+            {"name": "friend", "type": "User"},  # `null` is required
+        ],
+    }
+
+    records = [
+        {
+            "name": "juan",
+            "age": 20,
+            "friend": {
+                "name": "john",
+                "age": 30,
+            },
+        }
+    ]
+
+    with pytest.raises(ValueError, match="no value and no default for friend"):
+        roundtrip(schema, records)
+
+
 def test_enum_named_type():
     """https://github.com/fastavro/fastavro/issues/450"""
     schema = {