Recursive Schemas and Definitions

To create recursive or self-referencing data structures, you must use a combination of DefinitionsSchema to hold the schema definitions and DefinitionReferenceSchema to point to them. This pattern allows you to define a schema once and reference it multiple times, including within its own definition.

Creating a Recursive Schema

To implement recursion, wrap your main schema in definitions_schema and use definition_reference_schema to create the self-reference.

from pydantic_core import SchemaValidator, core_schema

# Define a recursive "Branch" structure
schema = core_schema.definitions_schema(
    # The main entry point schema
    core_schema.definition_reference_schema('Branch'),
    [
        # The actual definition of 'Branch'
        core_schema.typed_dict_schema(
            {
                'name': core_schema.typed_dict_field(core_schema.str_schema()),
                'sub_branch': core_schema.typed_dict_field(
                    core_schema.with_default_schema(
                        core_schema.nullable_schema(
                            # Self-reference back to 'Branch'
                            core_schema.definition_reference_schema('Branch')
                        ),
                        default=None,
                    )
                ),
            },
            ref='Branch',
        )
    ],
)

v = SchemaValidator(schema)

# Validating a nested structure
data = {'name': 'root', 'sub_branch': {'name': 'child', 'sub_branch': None}}
assert v.validate_python(data) == data

In this example:

• definitions_schema acts as the container for the entire schema logic.
• The definitions list contains a typed_dict_schema with a unique ref='Branch'.
• definition_reference_schema('Branch') is used both as the entry point and inside the fields to create the recursive loop.

Sharing Common Components

You can also use definitions to avoid duplicating complex schemas that are used in multiple places.

from pydantic_core import SchemaValidator, core_schema

# Define a shared integer schema with specific constraints
shared_int = core_schema.int_schema(ge=0, le=100, ref='percent_int')

schema = core_schema.definitions_schema(
    core_schema.typed_dict_schema({
        'math_score': core_schema.typed_dict_field(
            core_schema.definition_reference_schema('percent_int')
        ),
        'english_score': core_schema.typed_dict_field(
            core_schema.definition_reference_schema('percent_int')
        ),
    }),
    [shared_int]
)

v = SchemaValidator(schema)
assert v.validate_python({'math_score': 95, 'english_score': 88}) == {'math_score': 95, 'english_score': 88}

Customizing Serialization on References

You can override serialization behavior for a specific reference without changing the underlying definition. This is useful when you want a recursive structure to serialize differently at certain depths.

from pydantic_core import SchemaSerializer, core_schema

schema = core_schema.definitions_schema(
    core_schema.definition_reference_schema('Node'),
    [
        core_schema.typed_dict_schema(
            {
                'id': core_schema.typed_dict_field(core_schema.int_schema()),
                'next': core_schema.typed_dict_field(
                    core_schema.nullable_schema(
                        core_schema.definition_reference_schema(
                            'Node', 
                            # Force string serialization for the 'next' reference
                            serialization=core_schema.to_string_ser_schema(when_used='always')
                        )
                    )
                ),
            },
            ref='Node',
        )
    ],
)

s = SchemaSerializer(schema)
data = {'id': 1, 'next': {'id': 2, 'next': None}}
# The 'next' node is serialized as a string representation of the dict
result = s.to_python(data)
assert result['next'] == "{'id': 2, 'next': None}"

Troubleshooting

Missing References

If you use a definition_reference_schema with a schema_ref that does not exist in the definitions list, SchemaValidator will raise a SchemaError during initialization.

from pydantic_core import SchemaValidator, core_schema, SchemaError
import pytest

try:
    core_schema.definitions_schema(
        core_schema.definition_reference_schema('Missing'),
        [core_schema.int_schema(ref='Existing')]
    )
except SchemaError:
    # Raised because 'Missing' is not in the definitions list
    pass

Recursion Loops

Pydantic-core automatically detects cyclic references in your input data to prevent infinite loops. If a cycle is detected during validation, it raises a ValidationError with the type recursion_loop.

from pydantic_core import SchemaValidator, core_schema, ValidationError

# Using the Branch schema from the first example
v = SchemaValidator(...) 

cyclic_data = {'name': 'loop'}
cyclic_data['sub_branch'] = cyclic_data  # Create a cycle

try:
    v.validate_python(cyclic_data)
except ValidationError as e:
    assert e.errors()[0]['type'] == 'recursion_loop'

During serialization, detecting a circular reference will result in a ValueError with the message "Circular reference detected".